diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 0b1cefd57a..e5e6fc8683 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index fa31cfe1fb..410031ceec 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -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, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index 8f7e8e466a..29b39af969 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -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, diff --git a/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs b/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs index 0168b97b49..edbf872fbe 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs @@ -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::>(); + 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); } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 5dbcba6eef..caafcc01f0 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -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, diff --git a/codex-rs/codex-mcp/src/resource_origin.rs b/codex-rs/codex-mcp/src/resource_origin.rs index 0a21c4c63e..9783185310 100644 --- a/codex-rs/codex-mcp/src/resource_origin.rs +++ b/codex-rs/codex-mcp/src/resource_origin.rs @@ -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 { + (!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::>(); + 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 => { diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index 0aeff678ea..3dbfb460ed 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -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 { + 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, diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index f09a3293c1..bbd7d04fd8 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 777b1c882c..1fe8942b84 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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 diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index d801e10320..1d9848511d 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -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, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index f622dc0cf1..4a4d6dcfee 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -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 = diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 6f03b45952..dfde550e51 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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()), diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index f95a9d994e..23fc515d8e 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -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>, + pub mcp_resource_origins: Option, pub window_number: Option, pub first_window_id: Option, pub previous_window_id: Option, diff --git a/codex-rs/history/src/rollout_payload.rs b/codex-rs/history/src/rollout_payload.rs index 6d40c3d881..cc4621bf5a 100644 --- a/codex-rs/history/src/rollout_payload.rs +++ b/codex-rs/history/src/rollout_payload.rs @@ -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>>, #[serde(default, skip_serializing_if = "Option::is_none")] + mcp_resource_origins: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] window_number: Option, #[serde(default, skip_serializing_if = "Option::is_none")] first_window_id: Option>, @@ -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> 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), diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index 94f08c0ad8..fbd115fb38 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -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, diff --git a/codex-rs/protocol/src/mcp.rs b/codex-rs/protocol/src/mcp.rs index 7012d2b9f1..2d97dccd20 100644 --- a/codex-rs/protocol/src/mcp.rs +++ b/codex-rs/protocol/src/mcp.rs @@ -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, + pub turns: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub current_turn_id: Option, +} + +/// 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, + pub tool: String, + pub connector_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub link_id: Option, + 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]; diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index 58f8fbdeef..28a14eb2d6 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -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, diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 51b1fd69d9..7e83d8a74f 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -623,6 +623,7 @@ fn compacted(message: &str, replacement_history: Option>) -> 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, diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs index ceee507db8..9f9b0653e0 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs @@ -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( diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 741383267c..b188b35247 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -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) -> 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, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index d85dd3f7da..9a4a67916d 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -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, diff --git a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs index d3246a14e4..cb8bb4c5b0 100644 --- a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs +++ b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs @@ -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,