mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Expose conversation history to tool lifecycle extensions (#38403)
## What changed - Add a shared, read-only `ConversationHistorySnapshot` capability to the extension API. - Include the snapshot in `ToolStartInput`, preserving conversation order while excluding contextual user messages. - Avoid acquiring a history snapshot when no tool lifecycle contributors are registered. ## Testing - Verify snapshots share existing response items, remain stable as history changes, and filter contextual user messages. - Verify tool-start callbacks receive the history available for successive tool calls. GitOrigin-RevId: 822cae33781dd876ea628f104be08d8a449a4900
This commit is contained in:
@@ -9,6 +9,7 @@ use crate::event_mapping::is_contextual_user_message_content;
|
||||
use crate::session::turn_context::TurnContext;
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
|
||||
use codex_extension_api::ConversationHistorySnapshot;
|
||||
use codex_history::CodexHarnessMetadata;
|
||||
use codex_history::ResponseItemEnvelope;
|
||||
use codex_protocol::models::AgentMessageInputContent;
|
||||
@@ -63,6 +64,27 @@ pub(crate) struct ContextManager {
|
||||
world_state_baseline: Option<WorldStateSnapshot>,
|
||||
}
|
||||
|
||||
struct SharedConversationHistory {
|
||||
items: Arc<Vec<ResponseItemEnvelope>>,
|
||||
}
|
||||
|
||||
impl ConversationHistorySnapshot for SharedConversationHistory {
|
||||
fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
|
||||
Box::new(
|
||||
self.items
|
||||
.iter()
|
||||
.map(|envelope| &envelope.item)
|
||||
.filter(|item| {
|
||||
!matches!(
|
||||
item,
|
||||
ResponseItem::Message { role, content, .. }
|
||||
if role == "user" && is_contextual_user_message_content(content)
|
||||
)
|
||||
}),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl ContextManager {
|
||||
pub(crate) fn new() -> Self {
|
||||
Self {
|
||||
@@ -76,6 +98,12 @@ impl ContextManager {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn conversation_history_snapshot(&self) -> Arc<dyn ConversationHistorySnapshot> {
|
||||
Arc::new(SharedConversationHistory {
|
||||
items: Arc::clone(&self.items),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn token_info(&self) -> Option<TokenUsageInfo> {
|
||||
self.token_info.clone()
|
||||
}
|
||||
|
||||
@@ -112,6 +112,55 @@ fn raw_items(history: &ContextManager) -> Vec<ResponseItem> {
|
||||
history.raw_items().cloned().collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn conversation_history_snapshot_shares_response_items_until_history_changes() {
|
||||
let mut history = create_history_with_items(vec![assistant_msg("original")]);
|
||||
let snapshot = history.conversation_history_snapshot();
|
||||
|
||||
let original = history.raw_items().next().expect("original history item");
|
||||
let shared = snapshot.items().next().expect("shared snapshot item");
|
||||
assert!(std::ptr::eq(original, shared));
|
||||
|
||||
history.record_items(
|
||||
std::iter::once(&assistant_msg("later")),
|
||||
TruncationPolicy::Tokens(10_000),
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
snapshot.items().cloned().collect::<Vec<_>>(),
|
||||
vec![assistant_msg("original")],
|
||||
);
|
||||
assert_eq!(
|
||||
raw_items(&history),
|
||||
vec![assistant_msg("original"), assistant_msg("later")],
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn conversation_history_snapshot_excludes_contextual_user_messages() {
|
||||
let contextual_message = crate::context::ContextualUserFragment::into(UserInstructions {
|
||||
directory: None,
|
||||
text: "Follow the repository instructions.".to_string(),
|
||||
});
|
||||
let user_message = user_input_text_msg("Review this repository.");
|
||||
let assistant_message = assistant_msg("I will inspect the repository.");
|
||||
let developer_message = developer_msg(
|
||||
"# AGENTS.md instructions\n\n<INSTRUCTIONS>\nDeveloper context\n</INSTRUCTIONS>",
|
||||
);
|
||||
let history = create_history_with_items(vec![
|
||||
contextual_message,
|
||||
user_message.clone(),
|
||||
assistant_message.clone(),
|
||||
developer_message.clone(),
|
||||
]);
|
||||
let snapshot = history.conversation_history_snapshot();
|
||||
|
||||
assert_eq!(
|
||||
snapshot.items().cloned().collect::<Vec<_>>(),
|
||||
vec![user_message, assistant_message, developer_message],
|
||||
);
|
||||
}
|
||||
|
||||
struct TestWorldStateSection;
|
||||
|
||||
impl WorldStateSection for TestWorldStateSection {
|
||||
|
||||
@@ -56,6 +56,7 @@ use codex_connectors::connector_runtime_context_key;
|
||||
use codex_exec_server::Environment;
|
||||
use codex_exec_server::EnvironmentManager;
|
||||
use codex_execpolicy::prefix_rule_migration;
|
||||
use codex_extension_api::ConversationHistorySnapshot;
|
||||
use codex_extension_api::ExtensionDataInit;
|
||||
use codex_extension_api::LoadedUserInstructions;
|
||||
use codex_extension_api::PromptFragment;
|
||||
@@ -3668,6 +3669,13 @@ impl Session {
|
||||
state.clone_history()
|
||||
}
|
||||
|
||||
pub(crate) async fn conversation_history_snapshot(
|
||||
&self,
|
||||
) -> Arc<dyn ConversationHistorySnapshot> {
|
||||
let state = self.state.lock().await;
|
||||
state.history.conversation_history_snapshot()
|
||||
}
|
||||
|
||||
pub(crate) async fn current_window_id(&self) -> String {
|
||||
let state = self.state.lock().await;
|
||||
let thread_id = self.thread_id;
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use codex_extension_api::ToolCallOutcome;
|
||||
use codex_extension_api::ToolCallSource as ExtensionToolCallSource;
|
||||
use codex_extension_api::ToolFinishInput;
|
||||
@@ -10,20 +12,27 @@ use crate::tools::context::ToolCallSource;
|
||||
use crate::tools::context::ToolInvocation;
|
||||
|
||||
pub(crate) async fn notify_tool_start(invocation: &ToolInvocation) {
|
||||
for contributor in invocation
|
||||
let contributors = invocation
|
||||
.session
|
||||
.services
|
||||
.extensions
|
||||
.tool_lifecycle_contributors()
|
||||
{
|
||||
.tool_lifecycle_contributors();
|
||||
if contributors.is_empty() {
|
||||
return;
|
||||
}
|
||||
let thread_store = &invocation.session.services.thread_extension_data;
|
||||
let conversation_history = invocation.session.conversation_history_snapshot().await;
|
||||
|
||||
for contributor in contributors {
|
||||
contributor
|
||||
.on_tool_start(ToolStartInput {
|
||||
session_store: &invocation.session.services.session_extension_data,
|
||||
thread_store: &invocation.session.services.thread_extension_data,
|
||||
thread_store,
|
||||
turn_store: invocation.turn.extension_data.as_ref(),
|
||||
turn_id: invocation.turn.sub_id.as_str(),
|
||||
call_id: invocation.call_id.as_str(),
|
||||
tool_name: &invocation.tool_name,
|
||||
conversation_history: Arc::clone(&conversation_history),
|
||||
source: extension_tool_call_source(invocation.source.clone()),
|
||||
})
|
||||
.await;
|
||||
|
||||
@@ -145,6 +145,7 @@ mod stream_no_completed;
|
||||
mod subagent_notifications;
|
||||
mod token_budget;
|
||||
mod tool_harness;
|
||||
mod tool_lifecycle;
|
||||
mod tool_parallelism;
|
||||
mod tools;
|
||||
mod truncation;
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use anyhow::Result;
|
||||
use codex_core::config::Config;
|
||||
use codex_extension_api::ExtensionRegistryBuilder;
|
||||
use codex_extension_api::ResponseItem;
|
||||
use codex_extension_api::ToolLifecycleContributor;
|
||||
use codex_extension_api::ToolLifecycleFuture;
|
||||
use codex_extension_api::ToolStartInput;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use core_test_support::responses;
|
||||
use core_test_support::skip_if_no_network;
|
||||
use core_test_support::test_codex::test_codex;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
|
||||
struct RecordedHistory {
|
||||
call_id: String,
|
||||
items: Vec<ResponseItem>,
|
||||
}
|
||||
|
||||
struct ConversationHistoryRecorder {
|
||||
histories: Mutex<Vec<RecordedHistory>>,
|
||||
}
|
||||
|
||||
impl ToolLifecycleContributor for ConversationHistoryRecorder {
|
||||
fn on_tool_start<'a>(&'a self, input: ToolStartInput<'a>) -> ToolLifecycleFuture<'a> {
|
||||
Box::pin(async move {
|
||||
self.histories
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.push(RecordedHistory {
|
||||
call_id: input.call_id.to_owned(),
|
||||
items: input.conversation_history.items().cloned().collect(),
|
||||
});
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn tool_start_receives_conversation_history() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let server = responses::start_mock_server().await;
|
||||
let first_call_id = "first-plan-call";
|
||||
let second_call_id = "second-plan-call";
|
||||
responses::mount_sse_sequence(
|
||||
&server,
|
||||
vec![
|
||||
responses::sse(vec![
|
||||
responses::ev_reasoning_item("reasoning-1", &["inspect the workspace"], &[]),
|
||||
responses::ev_function_call(
|
||||
first_call_id,
|
||||
"update_plan",
|
||||
&json!({
|
||||
"plan": [{ "step": "Inspect workspace", "status": "in_progress" }]
|
||||
})
|
||||
.to_string(),
|
||||
),
|
||||
responses::ev_completed("first-response"),
|
||||
]),
|
||||
responses::sse(vec![
|
||||
responses::ev_function_call(
|
||||
second_call_id,
|
||||
"update_plan",
|
||||
&json!({
|
||||
"plan": [{ "step": "Inspect workspace", "status": "completed" }]
|
||||
})
|
||||
.to_string(),
|
||||
),
|
||||
responses::ev_completed("second-response"),
|
||||
]),
|
||||
responses::sse(vec![
|
||||
responses::ev_assistant_message("assistant-1", "done"),
|
||||
responses::ev_completed("third-response"),
|
||||
]),
|
||||
],
|
||||
)
|
||||
.await;
|
||||
|
||||
let recorder = Arc::new(ConversationHistoryRecorder {
|
||||
histories: Mutex::new(Vec::new()),
|
||||
});
|
||||
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
|
||||
extensions.tool_lifecycle_contributor(recorder.clone());
|
||||
let test = test_codex()
|
||||
.with_extensions(Arc::new(extensions.build()))
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
|
||||
let user_prompt = "Inspect the workspace and update the plan.";
|
||||
test.submit_text_turn(user_prompt).await?;
|
||||
|
||||
let histories = recorder
|
||||
.histories
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let call_ids = histories
|
||||
.iter()
|
||||
.map(|history| history.call_id.as_str())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(call_ids, vec![first_call_id, second_call_id]);
|
||||
|
||||
let first_history = &histories[0].items;
|
||||
assert!(first_history.iter().any(|item| matches!(
|
||||
item,
|
||||
ResponseItem::Message { role, content, .. }
|
||||
if role == "user"
|
||||
&& content.iter().any(|content| matches!(
|
||||
content,
|
||||
ContentItem::InputText { text } if text == user_prompt
|
||||
))
|
||||
)));
|
||||
assert!(
|
||||
first_history
|
||||
.iter()
|
||||
.any(|item| matches!(item, ResponseItem::Reasoning { .. }))
|
||||
);
|
||||
assert!(first_history.iter().any(|item| matches!(
|
||||
item,
|
||||
ResponseItem::FunctionCall { call_id, .. } if call_id == first_call_id
|
||||
)));
|
||||
|
||||
let second_history = &histories[1].items;
|
||||
assert!(second_history.iter().any(|item| matches!(
|
||||
item,
|
||||
ResponseItem::FunctionCallOutput { call_id, .. } if call_id == first_call_id
|
||||
)));
|
||||
assert!(second_history.iter().any(|item| matches!(
|
||||
item,
|
||||
ResponseItem::FunctionCall { call_id, .. } if call_id == second_call_id
|
||||
)));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
use codex_protocol::models::ResponseItem;
|
||||
|
||||
/// Read-only conversation-history snapshot supplied by the extension host.
|
||||
///
|
||||
/// Implementations should retain the host's existing snapshot storage rather than
|
||||
/// copying response payloads into an extension-owned collection.
|
||||
pub trait ConversationHistorySnapshot: Send + Sync {
|
||||
/// Returns the snapshot's response items in conversation order.
|
||||
fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_>;
|
||||
}
|
||||
@@ -1,10 +1,12 @@
|
||||
mod agent;
|
||||
mod conversation_history;
|
||||
mod events;
|
||||
mod metrics;
|
||||
mod response_items;
|
||||
|
||||
pub use agent::AgentSpawnFuture;
|
||||
pub use agent::AgentSpawner;
|
||||
pub use conversation_history::ConversationHistorySnapshot;
|
||||
pub use events::ExtensionEventSink;
|
||||
pub use events::ExtensionWarning;
|
||||
pub use events::NoopExtensionEventSink;
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
|
||||
use codex_tools::ToolName;
|
||||
|
||||
use crate::ConversationHistorySnapshot;
|
||||
use crate::ExtensionData;
|
||||
|
||||
/// Future returned by one tool-lifecycle callback.
|
||||
@@ -57,6 +59,8 @@ pub struct ToolStartInput<'a> {
|
||||
pub call_id: &'a str,
|
||||
/// Tool name as routed by the host.
|
||||
pub tool_name: &'a ToolName,
|
||||
/// Shared read-only snapshot of the conversation when the tool started.
|
||||
pub conversation_history: Arc<dyn ConversationHistorySnapshot>,
|
||||
/// Source that issued the tool call.
|
||||
pub source: ToolCallSource,
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ mod user_instructions;
|
||||
|
||||
pub use capabilities::AgentSpawnFuture;
|
||||
pub use capabilities::AgentSpawner;
|
||||
pub use capabilities::ConversationHistorySnapshot;
|
||||
pub use capabilities::ExtensionEventSink;
|
||||
pub use capabilities::ExtensionMetrics;
|
||||
pub use capabilities::ExtensionWarning;
|
||||
|
||||
Reference in New Issue
Block a user