mirror of
https://github.com/openai/codex.git
synced 2026-10-21 09:54:56 +00:00
Preserve MCP resource origins across compaction (#39192)
## Why Compaction can remove the tool-call events that associate an MCP app widget with the tool, account, and URI needed to authorize later resource reads. ## What changed - Store a bounded MCP resource-origin checkpoint with each compacted rollout. - Restore the checkpoint when resuming a session so existing widgets remain readable after compaction and restart. - Discard saved origins when rollback rewrites history across the compaction boundary, and reject oversized or invalid checkpoints during restoration. ## Testing - Extend the app-server MCP resource-origin test to compact a paginated thread, continue the conversation, and read the original widgets before and after a restart. - Verify rollback clears provenance that no longer matches retained history. GitOrigin-RevId: 7f91c6202fe8d9be21c78dda42fa205937c44263
This commit is contained in:
BIN
Binary file not shown.
@@ -3945,6 +3945,7 @@ mod tests {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -197,6 +197,7 @@ fn ignores_legacy_abort_without_turn_id_and_context_only_records() {
|
||||
let compacted = project(RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -10,6 +10,8 @@ use codex_app_server_protocol::McpResourceContent;
|
||||
use codex_app_server_protocol::McpResourceReadParams;
|
||||
use codex_app_server_protocol::McpResourceReadResponse;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ThreadCompactStartParams;
|
||||
use codex_app_server_protocol::ThreadCompactStartResponse;
|
||||
use codex_app_server_protocol::ThreadHistoryMode;
|
||||
use codex_app_server_protocol::ThreadResumeParams;
|
||||
use codex_app_server_protocol::ThreadResumeResponse;
|
||||
@@ -25,7 +27,8 @@ use std::sync::atomic::Ordering;
|
||||
use tokio::time::timeout;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn widget_reads_survive_history_modes_restarts_and_app_only_visibility() -> Result<()> {
|
||||
async fn widget_reads_survive_history_modes_compaction_restarts_and_app_only_visibility()
|
||||
-> Result<()> {
|
||||
let responses_server = responses::start_mock_server().await;
|
||||
let (apps_server_url, calls, apps_server_handle) = start_resource_apps_mcp_server().await?;
|
||||
calls.tools_enabled.store(true, Ordering::Relaxed);
|
||||
@@ -62,6 +65,22 @@ async fn widget_reads_survive_history_modes_restarts_and_app_only_visibility() -
|
||||
let mut model_responses = std::iter::repeat_n(response, 4)
|
||||
.flatten()
|
||||
.collect::<Vec<_>>();
|
||||
model_responses.insert(
|
||||
/*index*/ 4,
|
||||
responses::sse(vec![
|
||||
responses::ev_response_created("widget-compaction"),
|
||||
responses::ev_assistant_message("widget-summary", "The apps found matching lamps."),
|
||||
responses::ev_completed("widget-compaction"),
|
||||
]),
|
||||
);
|
||||
model_responses.insert(
|
||||
/*index*/ 5,
|
||||
responses::sse(vec![
|
||||
responses::ev_response_created("after-widget-compaction"),
|
||||
responses::ev_assistant_message("widget-follow-up", "The lamps are still available."),
|
||||
responses::ev_completed("after-widget-compaction"),
|
||||
]),
|
||||
);
|
||||
model_responses.push(responses::sse(vec![responses::ev_completed(
|
||||
"app-only-visibility",
|
||||
)]));
|
||||
@@ -161,6 +180,41 @@ async fn widget_reads_survive_history_modes_restarts_and_app_only_visibility() -
|
||||
);
|
||||
}
|
||||
if !ephemeral {
|
||||
if history_mode == ThreadHistoryMode::Paginated {
|
||||
let compact_id = app_server
|
||||
.send_thread_compact_start_request(ThreadCompactStartParams {
|
||||
thread_id: thread.id.clone(),
|
||||
})
|
||||
.await?;
|
||||
let _: ThreadCompactStartResponse =
|
||||
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(compact_id)).await??;
|
||||
timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
app_server.read_stream_until_notification_message("turn/completed"),
|
||||
)
|
||||
.await??;
|
||||
let turn_id = app_server
|
||||
.send_turn_start_request(TurnStartParams {
|
||||
thread_id: thread.id.clone(),
|
||||
input: vec![UserInput::Text {
|
||||
text: "What did the apps find?".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let _: TurnStartResponse =
|
||||
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(turn_id)).await??;
|
||||
timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
app_server.read_stream_until_notification_message("turn/completed"),
|
||||
)
|
||||
.await??;
|
||||
for call_id in ["walmart-call", "best-buy-call"] {
|
||||
let response = read_widget(&mut app_server, &thread.id, call_id).await?;
|
||||
assert_eq!(response.origin_call_id.as_deref(), Some(call_id));
|
||||
}
|
||||
}
|
||||
persistent_thread_ids.push(thread.id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -165,6 +165,7 @@ async fn thread_resume_paginated_model_context_preserves_original_metadata() ->
|
||||
&RolloutItem::Compacted(CompactedItem {
|
||||
message: "compacted history".to_string(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -8,6 +8,8 @@ use codex_connectors::AppToolPolicyInput;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::items::McpToolCallStatus;
|
||||
use codex_protocol::items::TurnItem;
|
||||
use codex_protocol::mcp::McpResourceOrigin;
|
||||
use codex_protocol::mcp::McpResourceOriginCheckpoint;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use rmcp::model::ReadResourceRequestParams;
|
||||
use rmcp::model::ReadResourceResult;
|
||||
@@ -37,6 +39,69 @@ pub(crate) struct ResourceOrigin {
|
||||
}
|
||||
|
||||
impl ResourceOrigins {
|
||||
pub(crate) fn checkpoint(&self) -> Option<McpResourceOriginCheckpoint> {
|
||||
(!self.origins.is_empty()).then(|| McpResourceOriginCheckpoint {
|
||||
origins: self
|
||||
.origins
|
||||
.iter()
|
||||
.map(|origin| McpResourceOrigin {
|
||||
call_id: origin.call_id.clone(),
|
||||
turn_id: origin.turn_id.clone(),
|
||||
tool: origin.tool.clone(),
|
||||
connector_id: origin.connector_id.clone(),
|
||||
link_id: origin.link_id.clone(),
|
||||
uri: origin.uri.clone(),
|
||||
ambiguous_account: origin.ambiguous_account,
|
||||
})
|
||||
.collect(),
|
||||
turns: self.turns.iter().cloned().collect(),
|
||||
current_turn_id: self.current_turn_id.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn restore_checkpoint(&mut self, checkpoint: &McpResourceOriginCheckpoint) {
|
||||
if checkpoint.origins.len() > MAX_ORIGINS
|
||||
|| checkpoint.turns.len() > MAX_ORIGINS
|
||||
|| checkpoint
|
||||
.turns
|
||||
.iter()
|
||||
.any(|turn_id| turn_id.len() > MAX_ORIGIN_BYTES)
|
||||
|| checkpoint
|
||||
.current_turn_id
|
||||
.as_ref()
|
||||
.is_some_and(|turn_id| turn_id.len() > MAX_ORIGIN_BYTES)
|
||||
{
|
||||
*self = Self::default();
|
||||
return;
|
||||
}
|
||||
|
||||
let origins = checkpoint
|
||||
.origins
|
||||
.iter()
|
||||
.map(|origin| ResourceOrigin {
|
||||
call_id: origin.call_id.clone(),
|
||||
turn_id: origin.turn_id.clone(),
|
||||
tool: origin.tool.clone(),
|
||||
connector_id: origin.connector_id.clone(),
|
||||
link_id: origin.link_id.clone(),
|
||||
uri: origin.uri.clone(),
|
||||
ambiguous_account: origin.ambiguous_account,
|
||||
})
|
||||
.collect::<VecDeque<_>>();
|
||||
if origins.iter().any(|origin| {
|
||||
origin.byte_len() > MAX_ORIGIN_BYTES || origin.connector_id.trim().is_empty()
|
||||
}) {
|
||||
*self = Self::default();
|
||||
return;
|
||||
}
|
||||
|
||||
*self = Self {
|
||||
origins,
|
||||
turns: checkpoint.turns.iter().cloned().collect(),
|
||||
current_turn_id: checkpoint.current_turn_id.clone(),
|
||||
};
|
||||
}
|
||||
|
||||
pub(crate) fn observe(&mut self, event: &EventMsg) {
|
||||
match event {
|
||||
EventMsg::TurnStarted(event) if event.turn_id.len() <= MAX_ORIGIN_BYTES => {
|
||||
|
||||
@@ -28,6 +28,7 @@ use codex_protocol::ThreadId;
|
||||
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
||||
use codex_protocol::mcp::CallToolResult;
|
||||
use codex_protocol::mcp::ClientMcpExtensions;
|
||||
use codex_protocol::mcp::McpResourceOriginCheckpoint;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::protocol::Event;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
@@ -196,6 +197,22 @@ impl McpRuntime {
|
||||
.observe(event);
|
||||
}
|
||||
|
||||
/// Captures bounded widget provenance for the next compaction checkpoint.
|
||||
pub fn resource_origin_checkpoint(&self) -> Option<McpResourceOriginCheckpoint> {
|
||||
self.resource_origins
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.checkpoint()
|
||||
}
|
||||
|
||||
/// Restores widget provenance retained by a compaction checkpoint.
|
||||
pub fn restore_resource_origin_checkpoint(&self, checkpoint: &McpResourceOriginCheckpoint) {
|
||||
self.resource_origins
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.restore_checkpoint(checkpoint);
|
||||
}
|
||||
|
||||
/// Reads a widget through the current binding of the app tool that produced it.
|
||||
pub async fn read_resource_for_call(
|
||||
&self,
|
||||
|
||||
@@ -1249,6 +1249,7 @@ async fn spawn_agent_numeric_fork_from_compacted_paginated_parent_clamps_to_prov
|
||||
}
|
||||
.into(),
|
||||
]),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1665,6 +1666,7 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
|
||||
replacement_history: Some(
|
||||
replacement_history.into_iter().map(Into::into).collect(),
|
||||
),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1839,6 +1841,7 @@ async fn spawn_agent_full_fork_restores_instructions_after_compaction_discards_p
|
||||
replacement_history: Some(
|
||||
replacement_history.into_iter().map(Into::into).collect(),
|
||||
),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1968,6 +1971,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "legacy compacted summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -3334,6 +3334,7 @@ impl Session {
|
||||
let compacted_item = CompactedItem {
|
||||
message: metadata.message,
|
||||
replacement_history: Some(items.clone()),
|
||||
mcp_resource_origins: self.services.mcp_runtime.resource_origin_checkpoint(),
|
||||
window_number: Some(metadata.window_number),
|
||||
first_window_id: Some(metadata.window_ids.first_window_id.to_string()),
|
||||
previous_window_id: metadata
|
||||
|
||||
@@ -1009,6 +1009,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1069,6 +1070,7 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1138,6 +1140,7 @@ async fn reconstruct_history_prefers_compacted_window_over_session_meta() {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(2),
|
||||
first_window_id: Some(compacted_first_window_id.to_string()),
|
||||
previous_window_id: Some(compacted_previous_window_id.to_string()),
|
||||
@@ -1173,6 +1176,7 @@ async fn reconstruct_history_replays_world_state_from_latest_compaction_window()
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1220,6 +1224,7 @@ async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "legacy summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1247,6 +1252,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "legacy summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1282,6 +1288,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "legacy summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1384,6 +1391,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1548,6 +1556,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1798,6 +1807,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1970,6 +1980,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -1224,8 +1224,20 @@ impl Session {
|
||||
mcp_projection.config.prefix_mcp_tool_names,
|
||||
));
|
||||
for item in initial_history.get_rollout_items() {
|
||||
if let RolloutItem::EventMsg(event) = item {
|
||||
mcp_runtime.observe_event(event);
|
||||
match item {
|
||||
RolloutItem::Compacted(compacted) => {
|
||||
if let Some(checkpoint) = &compacted.mcp_resource_origins {
|
||||
mcp_runtime.restore_resource_origin_checkpoint(checkpoint);
|
||||
}
|
||||
}
|
||||
RolloutItem::EventMsg(event) => mcp_runtime.observe_event(event),
|
||||
RolloutItem::SessionMeta(_)
|
||||
| RolloutItem::ResponseItem(_)
|
||||
| RolloutItem::InterAgentCommunication(_)
|
||||
| RolloutItem::InterAgentCommunicationMetadata { .. }
|
||||
| RolloutItem::TurnContext(_)
|
||||
| RolloutItem::WorldState(_)
|
||||
| RolloutItem::SecurityRiskScore(_) => {}
|
||||
}
|
||||
}
|
||||
let session_extension_data =
|
||||
|
||||
@@ -2052,6 +2052,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
|
||||
let rollout_items = vec![RolloutItem::Compacted(CompactedItem {
|
||||
message: String::new(),
|
||||
replacement_history: Some(replacement_history.clone()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(42),
|
||||
first_window_id: Some(first_window_id.to_string()),
|
||||
previous_window_id: Some(previous_window_id.to_string()),
|
||||
@@ -3893,6 +3894,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio
|
||||
.map(ResponseItemEnvelope::new)
|
||||
.collect(),
|
||||
),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(7),
|
||||
first_window_id: Some(first_window_id.to_string()),
|
||||
previous_window_id: Some(previous_window_id.to_string()),
|
||||
@@ -11167,6 +11169,7 @@ async fn sample_rollout(
|
||||
rollout_items.push(RolloutItem::Compacted(CompactedItem {
|
||||
message: summary1.to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(window_number),
|
||||
first_window_id: Some(window_ids.first_window_id.to_string()),
|
||||
previous_window_id: window_ids.previous_window_id.map(|id| id.to_string()),
|
||||
@@ -11214,6 +11217,7 @@ async fn sample_rollout(
|
||||
rollout_items.push(RolloutItem::Compacted(CompactedItem {
|
||||
message: summary2.to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(window_number),
|
||||
first_window_id: Some(window_ids.first_window_id.to_string()),
|
||||
previous_window_id: window_ids.previous_window_id.map(|id| id.to_string()),
|
||||
|
||||
@@ -9,6 +9,7 @@ use std::sync::Arc;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::capabilities::SelectedCapabilityRoot;
|
||||
use codex_protocol::dynamic_tools::DynamicToolSpec;
|
||||
use codex_protocol::mcp::McpResourceOriginCheckpoint;
|
||||
use codex_protocol::models::BaseInstructions;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
@@ -141,6 +142,7 @@ mod rollout_payload;
|
||||
pub struct CompactedItem {
|
||||
pub message: String,
|
||||
pub replacement_history: Option<Vec<ResponseItemEnvelope>>,
|
||||
pub mcp_resource_origins: Option<McpResourceOriginCheckpoint>,
|
||||
pub window_number: Option<u64>,
|
||||
pub first_window_id: Option<String>,
|
||||
pub previous_window_id: Option<String>,
|
||||
|
||||
@@ -4,6 +4,7 @@ use super::CodexHarnessMetadata;
|
||||
use super::CompactedItem;
|
||||
use super::EventMsg;
|
||||
use super::InterAgentCommunication;
|
||||
use super::McpResourceOriginCheckpoint;
|
||||
use super::ResponseItem;
|
||||
use super::ResponseItemEnvelope;
|
||||
use super::RolloutItem;
|
||||
@@ -131,6 +132,8 @@ pub(super) struct CompactedItemWire<'a> {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
replacement_history_metadata: Option<Vec<Cow<'a, CodexHarnessMetadata>>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
mcp_resource_origins: Option<Cow<'a, McpResourceOriginCheckpoint>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
window_number: Option<u64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
first_window_id: Option<Cow<'a, str>>,
|
||||
@@ -166,6 +169,7 @@ impl<'a> From<&'a CompactedItem> for CompactedItemWire<'a> {
|
||||
.collect()
|
||||
}),
|
||||
replacement_history_metadata,
|
||||
mcp_resource_origins: item.mcp_resource_origins.as_ref().map(Cow::Borrowed),
|
||||
window_number: item.window_number,
|
||||
first_window_id: item.first_window_id.as_deref().map(Cow::Borrowed),
|
||||
previous_window_id: item.previous_window_id.as_deref().map(Cow::Borrowed),
|
||||
@@ -229,6 +233,7 @@ impl TryFrom<CompactedItemWire<'_>> for CompactedItem {
|
||||
Ok(Self {
|
||||
message: item.message.into_owned(),
|
||||
replacement_history,
|
||||
mcp_resource_origins: item.mcp_resource_origins.map(Cow::into_owned),
|
||||
window_number,
|
||||
first_window_id: item.first_window_id.map(Cow::into_owned),
|
||||
previous_window_id: item.previous_window_id.map(Cow::into_owned),
|
||||
|
||||
@@ -188,6 +188,7 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
|
||||
},
|
||||
ResponseItemEnvelope::new(compaction_item.clone()),
|
||||
]),
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -289,6 +290,7 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() ->
|
||||
let compacted_line = serde_json::to_value(RolloutItem::Compacted(CompactedItem {
|
||||
message: "summary".to_string(),
|
||||
replacement_history: Some(vec![envelope]),
|
||||
mcp_resource_origins: Some(McpResourceOriginCheckpoint::default()),
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -430,6 +432,7 @@ fn compacted_item_serializes_window_number_and_id() -> Result<()> {
|
||||
let item = CompactedItem {
|
||||
message: "summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(3),
|
||||
first_window_id: Some("019b3f6e-0000-7000-8000-000000000001".to_string()),
|
||||
previous_window_id: Some("019b3f6e-0000-7000-8000-000000000002".to_string()),
|
||||
@@ -462,6 +465,7 @@ fn compacted_item_migrates_legacy_numeric_window_id() -> Result<()> {
|
||||
CompactedItem {
|
||||
message: "summary".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(3),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -16,6 +16,30 @@ pub const OPENAI_STANDARD_FORM_INPUT_EXTENSION_ID: &str = "openai/standard-form-
|
||||
/// Extension ID for MCP App UI rendering.
|
||||
pub const MCP_APP_UI_EXTENSION_ID: &str = "io.modelcontextprotocol/ui";
|
||||
|
||||
/// Bounded app-resource provenance retained across a compaction checkpoint.
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct McpResourceOriginCheckpoint {
|
||||
pub origins: Vec<McpResourceOrigin>,
|
||||
pub turns: Vec<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub current_turn_id: Option<String>,
|
||||
}
|
||||
|
||||
/// The original app, account, tool, and URI that authorize one widget read.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct McpResourceOrigin {
|
||||
pub call_id: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub turn_id: Option<String>,
|
||||
pub tool: String,
|
||||
pub connector_id: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub link_id: Option<String>,
|
||||
pub uri: String,
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub ambiguous_account: bool,
|
||||
}
|
||||
|
||||
/// Client extensions that must not be advertised to MCP servers.
|
||||
const MCP_CLIENT_ONLY_EXTENSION_IDS: [&str; 1] = [OPENAI_STANDARD_FORM_INPUT_EXTENSION_ID];
|
||||
|
||||
|
||||
@@ -207,6 +207,7 @@ fn builder_from_items_falls_back_to_filename() {
|
||||
let items = vec![RolloutItem::Compacted(CompactedItem {
|
||||
message: "noop".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -623,6 +623,7 @@ fn compacted(message: &str, replacement_history: Option<Vec<ResponseItem>>) -> R
|
||||
message: message.to_string(),
|
||||
replacement_history: replacement_history
|
||||
.map(|items| items.into_iter().map(Into::into).collect()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -240,6 +240,7 @@ impl RollbackPlanner {
|
||||
.filter_map(|mut frame| {
|
||||
if Some(frame.record_index) == replay_anchor {
|
||||
frame.item.replacement_history = Some(Vec::new());
|
||||
frame.item.mcp_resource_origins = None;
|
||||
return Some((frame.record_index, frame.item));
|
||||
}
|
||||
frame
|
||||
@@ -316,6 +317,7 @@ impl RollbackPlanner {
|
||||
let post_compaction_turns = depth_before.saturating_sub(frame.boundary_depth);
|
||||
let remaining = count.saturating_sub(post_compaction_turns);
|
||||
if remaining > 0 {
|
||||
frame.item.mcp_resource_origins = None;
|
||||
let replacement_history =
|
||||
frame.item.replacement_history.as_mut().ok_or_else(|| {
|
||||
migration_error(
|
||||
|
||||
@@ -15,6 +15,8 @@ use codex_protocol::ThreadId;
|
||||
use codex_protocol::config_types::ReasoningSummary;
|
||||
use codex_protocol::items::ReasoningItem;
|
||||
use codex_protocol::items::TurnItem;
|
||||
use codex_protocol::mcp::McpResourceOrigin;
|
||||
use codex_protocol::mcp::McpResourceOriginCheckpoint;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::protocol::AgentMessageEvent;
|
||||
@@ -220,6 +222,7 @@ fn compacted(replacement_history: Vec<ResponseItem>) -> RolloutItem {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "checkpoint".to_string(),
|
||||
replacement_history: Some(replacement_history.into_iter().map(Into::into).collect()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1019,6 +1022,33 @@ async fn migration_rolls_back_inter_agent_metadata_with_its_delivery() {
|
||||
async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
|
||||
let home = TempDir::new().expect("create Codex home");
|
||||
let thread_id = ThreadId::new();
|
||||
let RolloutItem::Compacted(mut checkpoint) = compacted(vec![
|
||||
input_response_message("user", "old question"),
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "assistant".to_string(),
|
||||
content: vec![ContentItem::OutputText {
|
||||
text: "old answer".to_string(),
|
||||
}],
|
||||
phase: None,
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
},
|
||||
]) else {
|
||||
unreachable!("compacted helper always creates a compaction checkpoint");
|
||||
};
|
||||
checkpoint.mcp_resource_origins = Some(McpResourceOriginCheckpoint {
|
||||
origins: vec![McpResourceOrigin {
|
||||
call_id: "widget-call".to_string(),
|
||||
turn_id: Some("keep-before-compaction".to_string()),
|
||||
tool: "_product_search".to_string(),
|
||||
connector_id: "shopping".to_string(),
|
||||
link_id: None,
|
||||
uri: "ui://shopping/widget".to_string(),
|
||||
ambiguous_account: false,
|
||||
}],
|
||||
turns: vec!["keep-before-compaction".to_string()],
|
||||
current_turn_id: Some("keep-before-compaction".to_string()),
|
||||
});
|
||||
let path = write_rollout(
|
||||
home.path(),
|
||||
thread_id,
|
||||
@@ -1027,18 +1057,7 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
|
||||
started("keep-before-compaction"),
|
||||
user_message("old question"),
|
||||
completed("keep-before-compaction"),
|
||||
compacted(vec![
|
||||
input_response_message("user", "old question"),
|
||||
ResponseItem::Message {
|
||||
id: None,
|
||||
role: "assistant".to_string(),
|
||||
content: vec![ContentItem::OutputText {
|
||||
text: "old answer".to_string(),
|
||||
}],
|
||||
phase: None,
|
||||
internal_chat_message_metadata_passthrough: None,
|
||||
},
|
||||
]),
|
||||
RolloutItem::Compacted(checkpoint),
|
||||
started("remove-after-compaction"),
|
||||
user_message("new question"),
|
||||
completed("remove-after-compaction"),
|
||||
@@ -1067,14 +1086,15 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
|
||||
.await
|
||||
.expect("read rollback-through-compaction turns");
|
||||
assert_eq!(turns.turns.len(), 1);
|
||||
let replacement_history = read_rollout(&path)
|
||||
let checkpoint = read_rollout(&path)
|
||||
.into_iter()
|
||||
.find_map(|line| match line.item {
|
||||
RolloutItem::Compacted(item) => item.replacement_history,
|
||||
RolloutItem::Compacted(item) => Some(item),
|
||||
_ => None,
|
||||
})
|
||||
.expect("retained compaction");
|
||||
assert!(replacement_history.is_empty());
|
||||
assert_eq!(checkpoint.replacement_history, Some(Vec::new()));
|
||||
assert_eq!(checkpoint.mcp_resource_origins, None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1363,6 +1383,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "superseded checkpoint".repeat(1024),
|
||||
replacement_history: Some(Vec::new()),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -1382,6 +1403,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
|
||||
}
|
||||
.into(),
|
||||
]),
|
||||
mcp_resource_origins: None,
|
||||
window_number: Some(2),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -581,6 +581,7 @@ mod tests {
|
||||
let item = RolloutItem::Compacted(CompactedItem {
|
||||
message: "compacted".to_string(),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
@@ -72,6 +72,7 @@ fn legacy_transcript_preview_scans_tail_across_compaction() {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: String::from("summary that is not transcript text"),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -202,6 +203,7 @@ fn legacy_transcript_preview_falls_back_for_oversized_hidden_record() {
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
@@ -228,6 +230,7 @@ fn legacy_transcript_preview_falls_back_when_scan_budget_is_exhausted() {
|
||||
let compacted = RolloutItem::Compacted(CompactedItem {
|
||||
message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES / 8),
|
||||
replacement_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
|
||||
Reference in New Issue
Block a user