Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,11 @@ async fn app_server_uses_configured_notes_backend_for_context_window_hints(
})
.count();
assert_eq!(bridge_calls, usize::from(!use_history_notes_extension));
let developer_messages = request.message_input_texts("developer");
let developer_messages: Vec<_> = request
.message_input_text_groups("developer")
.into_iter()
.map(|parts| parts.concat())
.collect();
let context_window = developer_messages
.iter()
.find(|text| text.contains("<context_window>"))
Expand Down
2 changes: 1 addition & 1 deletion codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ pub fn tool_is_model_visible(tool: &ToolInfo) -> bool {
}

/// Catalog identity within one published connection set, including cached declarations.
#[derive(PartialEq, Eq)]
#[derive(Clone, PartialEq, Eq)]
pub(crate) enum BindingCatalogRevision {
Ready(ClientToolCatalogRevision),
Cached(u64),
Expand Down
1 change: 1 addition & 0 deletions codex-rs/codex-mcp/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ pub use runtime::McpRuntime;
pub use runtime::McpRuntimeContext;
pub use runtime::McpRuntimeInput;
pub use runtime::McpStartupPolicy;
pub use runtime::McpToolCallResult;
pub use runtime::apply_http_headers_helper;
pub use sandbox_metadata::MCP_SANDBOX_LAUNCHER_META_KEY;
pub use sandbox_metadata::MCP_SANDBOX_STATE_CAPABILITY;
Expand Down
70 changes: 57 additions & 13 deletions codex-rs/codex-mcp/src/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,14 @@ pub enum McpStartupPolicy {
LazyWhenCached,
}

/// A host-requested tool result and its identity in the published MCP catalog.
pub struct McpToolCallResult {
pub result: CallToolResult,
/// None when the cached catalog is missing, changed, or has no entry for the tool.
/// This is not MCP wire metadata.
pub source_tool_namespace: Option<String>,
}

/// Configuration and owning-thread state needed to materialize an MCP runtime.
pub struct McpRuntimeInput {
pub startup_policy: McpStartupPolicy,
Expand Down Expand Up @@ -639,6 +647,19 @@ impl McpRuntime {
self.latest_connections().list_all_tools().await
}

pub async fn latest_read_resource(
&self,
server: &str,
params: ReadResourceRequestParams,
) -> anyhow::Result<ReadResourceResult> {
self.latest_connections()
.read_resource(server, params)
.await
}

/// Calls a tool for the host, using a cached namespace only if its catalogs stay unchanged.
/// Missing or changed catalogs leave the namespace unknown. Attribution never starts
/// servers or refreshes their catalogs.
#[allow(clippy::too_many_arguments)]
pub async fn latest_call_tool(
&self,
Expand All @@ -649,8 +670,16 @@ impl McpRuntime {
meta: Option<serde_json::Value>,
requested_timeout: Option<Duration>,
wait_for_server: bool,
) -> anyhow::Result<CallToolResult> {
self.latest_connections()
) -> anyhow::Result<McpToolCallResult> {
let current = self.current.load_full();
let cached = current
.cached_binding
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.and_then(|cached| Some((cached.catalog_revisions.clone(), cached.binding.upgrade()?)));
let result = current
.connections
.call_tool(
server,
tool,
Expand All @@ -660,17 +689,32 @@ impl McpRuntime {
requested_timeout,
wait_for_server,
)
.await
}

pub async fn latest_read_resource(
&self,
server: &str,
params: ReadResourceRequestParams,
) -> anyhow::Result<ReadResourceResult> {
self.latest_connections()
.read_resource(server, params)
.await
.await?;
// Host calls use live connections, unlike model calls pinned to a binding.
// Check after dispatch so a refresh before or during the call invalidates attribution.
// Revisions include client identity and increase on catalog changes. Reading them
// does not start servers or refresh catalogs; all catalogs matter for name collisions.
let namespace = if let Some((revisions, binding)) = cached
&& current
.connections
.stable_catalog_revisions(
/*required_servers*/ &[],
/*required_plugins*/ &HashSet::new(),
)
.await
.as_ref()
== Some(&revisions)
{
binding
.tool_info(server, tool)
.map(|info| info.callable_namespace.clone())
} else {
None
};
Ok(McpToolCallResult {
result,
source_tool_namespace: namespace,
})
}

pub async fn latest_wait_for_server_ready(&self, server: &str, timeout: Duration) -> bool {
Expand Down
169 changes: 150 additions & 19 deletions codex-rs/context-fragments/src/annotated_content.rs
Original file line number Diff line number Diff line change
@@ -1,23 +1,95 @@
//! Keep content and its positional annotations together during history edits.

use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::ContentItemMetadata;
use codex_protocol::models::ResponseItem;
use codex_utils_string::plan_token_truncation;

/// Model-visible content paired with its harness-owned classification.
/// Content, classification and attribution travel together through edits and truncation.
#[derive(Clone, Debug, PartialEq)]
pub struct AnnotatedContent {
content: ContentItem,
kind: ContentItemKind,
metadata: ContentItemMetadata,
}

/// Applies the string truncator's head/tail selection to the combined text.
/// Retained slices keep their kind and source; the generated notice has harness
/// provenance. Non-text parts are retained unchanged, without cloning media.
pub fn truncate_annotated_text(
parts: Vec<AnnotatedContent>,
max_tokens: usize,
) -> Vec<AnnotatedContent> {
let text: String = parts
.iter()
.filter_map(|part| match part.content() {
ContentItem::InputText { text } | ContentItem::OutputText { text, .. } => {
Some(text.as_str())
}
ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => None,
})
.collect();
let Some(truncation) = plan_token_truncation(&text, max_tokens) else {
return parts;
};
let mut notice = Some(AnnotatedContent::text(
truncation.marker,
ContentItemKind("unknown".to_string()),
ContentItemMetadata::harness(),
));
let mut result = Vec::with_capacity(parts.len() + 2);
let mut offset = 0;
for part in parts {
let length = match part.content() {
ContentItem::InputText { text } | ContentItem::OutputText { text, .. } => text.len(),
ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => {
result.push(part);
continue;
}
};
let start = offset;
offset += length;
if offset <= truncation.removed.start || start >= truncation.removed.end {
result.push(part);
continue;
}
let prefix_end = truncation.removed.start.saturating_sub(start);
let suffix_start = truncation.removed.end.saturating_sub(start).min(length);
if prefix_end > 0 {
result.push(part.text_slice(0..prefix_end));
}
if let Some(notice) = notice.take() {
result.push(notice);
}
if suffix_start < length {
result.push(part.text_slice(suffix_start..length));
}
}
result
}

impl AnnotatedContent {
/// Creates content and its classification together.
pub fn new(content: ContentItem, kind: ContentItemKind) -> Self {
Self { content, kind }
Self {
content,
kind,
metadata: ContentItemMetadata::default(),
}
}

/// Creates model-visible input text and its classification together.
pub fn input_text(text: impl Into<String>, kind: ContentItemKind) -> Self {
Self::new(ContentItem::InputText { text: text.into() }, kind)
/// Creates text with the attribution supplied by its producer.
pub fn text(
text: impl Into<String>,
kind: ContentItemKind,
metadata: ContentItemMetadata,
) -> Self {
Self {
content: ContentItem::InputText { text: text.into() },
kind,
metadata,
}
}

/// Returns the model-visible content.
Expand All @@ -35,16 +107,54 @@ impl AnnotatedContent {
&self.kind
}

/// Separates the content from its classification at an API boundary.
pub fn into_parts(self) -> (ContentItem, ContentItemKind) {
(self.content, self.kind)
/// Separates content and its positional annotations at an API boundary.
pub fn into_parts(self) -> (ContentItem, ContentItemKind, ContentItemMetadata) {
(self.content, self.kind, self.metadata)
}

// Called only for the UTF-8-aligned text ranges selected by the truncator.
fn text_slice(&self, range: std::ops::Range<usize>) -> Self {
let content = match &self.content {
ContentItem::InputText { text } => ContentItem::InputText {
text: text[range].to_owned(),
},
ContentItem::OutputText { text, .. } => ContentItem::OutputText {
text: text[range].to_owned(),
// Original offsets and token probabilities do not describe a shortened slice.
annotations: None,
logprobs: None,
},
ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => {
unreachable!("only text parts have byte ranges")
}
};
Self {
content,
kind: self.kind.clone(),
metadata: self.metadata.clone(),
}
}
}

/// Takes a message's content together with its positional classifications.
/// Assembles a message without separating its content from its attribution.
pub fn message_from_parts(role: impl Into<String>, parts: Vec<AnnotatedContent>) -> ResponseItem {
let mut item = ResponseItem::Message {
status: None,
encrypted_content: None,
id: None,
role: role.into(),
content: Vec::new(),
phase: None,
internal_chat_message_metadata_passthrough: None,
};
set_annotated_content(&mut item, parts);
item
}

/// Takes a message's content together with its classification and source metadata.
///
/// Legacy messages, including persisted rollouts, may not have classifications.
/// Missing entries are classified as unknown so the message remains usable.
/// Missing entries have unknown classification and source so the message remains usable.
pub fn to_annotated_content(item: &mut ResponseItem) -> Option<Vec<AnnotatedContent>> {
let ResponseItem::Message {
content,
Expand All @@ -59,19 +169,32 @@ pub fn to_annotated_content(item: &mut ResponseItem) -> Option<Vec<AnnotatedCont
.as_mut()
.and_then(|metadata| metadata.content_item_kinds.take())
.unwrap_or_default();
let metadata = internal_chat_message_metadata_passthrough
.as_mut()
.and_then(|metadata| metadata.content_item_metadata.take())
.unwrap_or_default();

Some(
std::mem::take(content)
.into_iter()
.zip(kinds.into_iter().chain(std::iter::repeat_with(|| {
ContentItemKind("unknown".to_string())
})))
.map(|(content, kind)| AnnotatedContent::new(content, kind))
.zip(
metadata
.into_iter()
.chain(std::iter::repeat_with(ContentItemMetadata::default)),
)
.map(|((content, kind), metadata)| AnnotatedContent {
content,
kind,
metadata,
})
.collect(),
)
}

/// Replaces a message's content and positional classifications together.
/// Replaces content and its positional annotations together.
pub fn set_annotated_content(
item: &mut ResponseItem,
annotated_content: Vec<AnnotatedContent>,
Expand All @@ -85,14 +208,22 @@ pub fn set_annotated_content(
return None;
};

let (updated_content, content_item_kinds): (Vec<_>, Vec<_>) = annotated_content
.into_iter()
.map(AnnotatedContent::into_parts)
.unzip();
let mut updated_content = Vec::with_capacity(annotated_content.len());
let mut content_item_kinds = Vec::with_capacity(annotated_content.len());
let mut content_item_metadata = Vec::with_capacity(annotated_content.len());
for annotated in annotated_content {
let (content, kind, metadata) = annotated.into_parts();
updated_content.push(content);
content_item_kinds.push(kind);
content_item_metadata.push(metadata);
}
*content = updated_content;
internal_chat_message_metadata_passthrough
.get_or_insert_default()
.content_item_kinds = Some(content_item_kinds);
let metadata = internal_chat_message_metadata_passthrough.get_or_insert_default();
metadata.content_item_kinds = Some(content_item_kinds);
metadata.content_item_metadata = content_item_metadata
.iter()
.any(|metadata| metadata != &ContentItemMetadata::default())
.then_some(content_item_metadata);

Some(())
}
Loading
Loading