Defer extension tool history materialization until first access (#47035)

## Why

Extension tool calls eagerly copied and converted conversation history even when the tool never read it.

## What changed

Retain a shared invocation-time snapshot and defer conversion to raw response items until `ConversationHistory::items()` is called. Cache the result across clones with `LazyLock`, and keep debug formatting from triggering materialization. Release unrelated history metadata when capturing the shared items.

## Testing

Extend the extension executor test to verify that history first read after a later turn still reflects the invocation-time snapshot, in both legacy and paginated history modes. Update the history API test to check that shared items preserve their envelopes.

GitOrigin-RevId: c737b765b7e4889735dc78e12d685d36e854570b
This commit is contained in:
Charlie Marsh
2026-09-21 13:56:17 +00:00
committed by copyberry
parent f585f13a12
commit 3ead673946
5 changed files with 89 additions and 21 deletions
+6 -9
View File
@@ -496,17 +496,14 @@ impl ContextManager {
&self.items
}
/// Returns raw items in the history and consumes the snapshot.
pub(crate) fn into_raw_items(self) -> Vec<ResponseItem> {
self.into_annotated_items()
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
}
/// Returns annotated history items and consumes the snapshot.
pub(crate) fn into_annotated_items(self) -> Vec<ResponseItemEnvelope> {
Arc::unwrap_or_clone(self.items)
Arc::unwrap_or_clone(self.into_shared_annotated_items())
}
/// Keeps shared response items while releasing the snapshot's unrelated metadata.
pub(crate) fn into_shared_annotated_items(self) -> Arc<Vec<ResponseItemEnvelope>> {
self.items
}
pub(crate) fn history_version(&self) -> u64 {
@@ -765,9 +765,8 @@ fn cloned_history_shares_items_until_mutated() {
#[test]
fn annotated_history_apis_preserve_envelopes() {
let first_item = assistant_msg("first");
let first_envelope = ResponseItemEnvelope {
item: first_item.clone(),
item: assistant_msg("first"),
metadata: Some(CodexHarnessMetadata::default()),
};
let mut history = ContextManager::new();
@@ -778,7 +777,10 @@ fn annotated_history_apis_preserve_envelopes() {
history.annotated_items(),
std::slice::from_ref(&first_envelope)
);
assert_eq!(history.into_raw_items(), vec![first_item]);
assert_eq!(
history.into_shared_annotated_items().as_slice(),
&[first_envelope]
);
}
#[test_case(None, 100, 5, true; "model policy")]
@@ -2,6 +2,7 @@ use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::Weak;
use codex_history::ResponseItemEnvelope;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -164,8 +165,17 @@ impl TurnItemEmitter for CoreTurnItemEmitter {
}
async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall<'_> {
let conversation_history =
ConversationHistory::new(invocation.session.clone_history().await.into_raw_items());
let history = invocation
.session
.clone_history()
.await
.into_shared_annotated_items();
let conversation_history = ConversationHistory::new_deferred(move || {
Arc::unwrap_or_clone(history)
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
});
let settings = &invocation.step_context.settings;
let codex_turn_metadata = invocation
.turn
+39 -4
View File
@@ -30,6 +30,7 @@ use codex_protocol::dynamic_tools::DynamicToolFunctionSpec;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ApplyPatchToolType;
@@ -57,6 +58,7 @@ use codex_protocol::protocol::CONTEXT_WINDOW_GUIDANCE_OPEN_TAG;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SafetyBufferingEvent;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::TurnSettingsUpdate;
use codex_protocol::protocol::TurnSettingsUpdateOutcome;
@@ -64,6 +66,7 @@ use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputEvent;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_protocol::user_input::UserInput;
use codex_tools::ConversationHistory;
use codex_tools::JsonToolOutput;
use codex_tools::ToolCall;
use codex_tools::ToolExecutor;
@@ -106,6 +109,7 @@ use serde_json::Value;
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use test_case::test_case;
use super::rmcp_client::remote_aware_environment_id;
@@ -2898,7 +2902,8 @@ async fn captured_step_controls_mcp_output_limit(supports_images: bool) -> Resul
Ok(())
}
struct SettingsEcho;
#[derive(Clone)]
struct SettingsEcho(Arc<Mutex<Option<ConversationHistory>>>);
impl ToolContributor for SettingsEcho {
fn tools(
@@ -2906,7 +2911,7 @@ impl ToolContributor for SettingsEcho {
_session_store: &ExtensionData,
_thread_store: &ExtensionData,
) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
vec![Arc::new(Self)]
vec![Arc::new(self.clone())]
}
}
@@ -2931,6 +2936,7 @@ impl<'call> ToolExecutor<ToolCall<'call>> for SettingsEcho {
'call: 'a,
{
Box::pin(async move {
*self.0.lock().expect("history lock") = Some(call.conversation_history);
let metadata: Value = serde_json::from_str(
call.codex_turn_metadata
.as_deref()
@@ -2947,8 +2953,12 @@ impl<'call> ToolExecutor<ToolCall<'call>> for SettingsEcho {
}
}
#[test_case(ThreadHistoryMode::Legacy; "legacy")]
#[test_case(ThreadHistoryMode::Paginated; "paginated")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn captured_step_settings_reach_extension_executor() -> Result<()> {
async fn captured_step_settings_and_history_reach_extension_executor(
history_mode: ThreadHistoryMode,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let responses = mount_sse_sequence(
@@ -2961,12 +2971,15 @@ async fn captured_step_settings_reach_extension_executor() -> Result<()> {
ev_completed("resp-b"),
]),
sse_completed("resp-result"),
sse_completed("resp-later"),
],
)
.await;
let history = Arc::default();
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
extensions.tool_contributor(Arc::new(SettingsEcho));
extensions.tool_contributor(Arc::new(SettingsEcho(Arc::clone(&history))));
let test = direct_tool_settings_test()
.with_history_mode(history_mode)
.with_extensions(Arc::new(extensions.build()))
.with_config(|config| {
for model in &mut config.model_catalog.as_mut().expect("models").models {
@@ -3004,6 +3017,28 @@ async fn captured_step_settings_reach_extension_executor() -> Result<()> {
"output_bytes": 512,
})
);
test.submit_text_turn("later turn").await?;
// First read after a later turn: the extension must retain its invocation-time snapshot.
let history = history
.lock()
.expect("history lock")
.take()
.expect("captured history");
let user_texts = history
.items()
.iter()
.filter_map(|item| match item {
ResponseItem::Message { role, content, .. } if role == "user" => Some(content),
_ => None,
})
.flatten()
.filter_map(|item| match item {
ContentItem::InputText { text } => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>();
assert!(user_texts.contains(&"pause before continuing"));
assert!(!user_texts.contains(&"later turn"));
Ok(())
}
+27 -3
View File
@@ -9,21 +9,31 @@ use codex_protocol::protocol::EventMsg;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_output_truncation::TruncationPolicy;
use codex_utils_output_truncation::with_serialization_allowance;
use std::fmt;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::LazyLock;
/// Raw response history snapshot available when an extension tool is invoked.
#[derive(Clone, Debug, Default)]
#[derive(Clone)]
pub struct ConversationHistory {
items: Arc<[ResponseItem]>,
items: Arc<LazyLock<Box<[ResponseItem]>, HistoryLoader>>,
}
type HistoryLoader = Box<dyn FnOnce() -> Box<[ResponseItem]> + Send + Sync>;
impl ConversationHistory {
pub fn new(items: Vec<ResponseItem>) -> Self {
Self::new_deferred(move || items)
}
/// Materializes an invocation-time snapshot only when a tool reads its history.
/// The loader must capture that snapshot rather than querying mutable session state.
pub fn new_deferred(load: impl FnOnce() -> Vec<ResponseItem> + Send + Sync + 'static) -> Self {
Self {
items: items.into(),
items: Arc::new(LazyLock::new(Box::new(move || load().into_boxed_slice()))),
}
}
@@ -32,6 +42,20 @@ impl ConversationHistory {
}
}
impl Default for ConversationHistory {
fn default() -> Self {
Self::new(Vec::new())
}
}
impl fmt::Debug for ConversationHistory {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ConversationHistory")
.finish_non_exhaustive()
}
}
/// Future returned when an extension tool emits a visible turn-item lifecycle event.
pub type TurnItemEmissionFuture<'a> = Pin<Box<dyn Future<Output = ()> + Send + 'a>>;