github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add diagnostic reasons to MCP attribution errors (#47899)

Browse Source 
## Why

The `attribution_error` status does not explain why MCP attribution is incomplete. Preserve a diagnostic reason so callers can distinguish history and recorder errors from request serialization failures.

## What changed

- Add optional `error_reason` diagnostics to `McpAttribution`, retaining the first recorder error through checkpoints, compaction, resume, and fork.
- Identify missing or invalid checkpoints, conflicting sources, invalid sources, and poisoned recorder state. Use `restored_error_unknown` for older error checkpoints without a reason.
- Distinguish `payload_too_large` from `serialization_failed` in request metadata fallbacks.
- Tolerate missing, unknown, or malformed diagnostic reasons when reading history, without letting diagnostics determine attribution status.

## Testing

Add coverage for reason preservation, recorder poisoning, backward-compatible checkpoint decoding, resume and cold-fork restoration, and request metadata serialization, including oversized payloads.

GitOrigin-RevId: 389b9446fae119f793978f3e7e4a926677e84d8c
This commit is contained in:
peilin-openai
2026-09-24 12:02:00 -07:00
committed by Ian MacLeod
parent b7add4df3d
commit 0d7f10b874
10 changed files with 286 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/core/src/agent/control_tests.rs
+23 -2
View File
@@ -59,6 +59,7 @@ use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::mcp::McpAttribution;
use codex_protocol::mcp::McpAttributionErrorReason;
use codex_protocol::mcp::McpAttributionSource;
use codex_protocol::mcp::McpAttributionStatus;
use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID;
@@ -3159,8 +3160,12 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc
}
}
#[test_case::test_case(false; "complete")]
#[test_case::test_case(true; "error_reason")]
#[tokio::test]
async fn resume_and_cold_fork_restore_mcp_attribution_in_constructed_requests() {
async fn resume_and_cold_fork_restore_mcp_attribution_in_constructed_requests(
record_invalid_source: bool,
) {
let harness = AgentControlHarness::new().await;
let (source_thread_id, source_thread) = harness.start_thread().await;
let turn_context = source_thread.session.new_default_turn().await;
@@ -3184,6 +3189,16 @@ async fn resume_and_cold_fork_restore_mcp_attribution_in_constructed_requests()
.services
.executed_tool_calls
.record_mcp_source(source.clone());
if record_invalid_source {
source_thread
.session
.services
.executed_tool_calls
.record_mcp_source(McpAttributionSource {
tool_name: String::new(),
..source.clone()
});
}
source_thread
.session
.record_conversation_items(
@@ -3232,7 +3247,12 @@ async fn resume_and_cold_fork_restore_mcp_attribution_in_constructed_requests()
.await
.expect("resume source thread");
let expected = McpAttribution {
status: McpAttributionStatus::Complete,
status: if record_invalid_source {
McpAttributionStatus::AttributionError
} else {
McpAttributionStatus::Complete
},
error_reason: record_invalid_source.then_some(McpAttributionErrorReason::SourceInvalid),
sources: vec![source],
};
assert_eq!(
@@ -3475,6 +3495,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() {
mcp_attribution_in_constructed_request(&child_thread).await,
McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source],
}
);
codex-rs/core/src/client_tests.rs
+32 -1
View File
@@ -499,6 +499,7 @@ fn responses_request_limits_internal_metadata_to_resolved_first_party_https_endp
let without_raw_metadata = output_with_tool_result_metadata(ToolResultMetadata::default());
let attribution = McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![McpAttributionSource {
connector_id: Some("connector_example".to_string()),
plugin_id: Some("example@openai-bundled".to_string()),
@@ -585,6 +586,36 @@ fn responses_request_limits_internal_metadata_to_resolved_first_party_https_endp
assert_eq!(prompt.input, vec![output.clone()]);
}
}
responses_metadata.mcp_attribution = Some(McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(
codex_protocol::mcp::McpAttributionErrorReason::HistoryMissingCheckpoint,
),
sources: attribution.sources.clone(),
});
let request = client.build_responses_request(
&prompt,
&test_model_info(),
/*effort*/ None,
codex_protocol::config_types::ReasoningSummary::None,
/*service_tier*/ None,
&responses_metadata,
/*include_internal*/ true,
)?;
assert_eq!(
request
.client_metadata
.as_ref()
.and_then(|metadata| metadata.get(MCP_ATTRIBUTION_CLIENT_METADATA_KEY))
.map(|value| serde_json::from_str::<serde_json::Value>(value))
.transpose()?,
Some(json!({
"status": "attribution_error",
"sources": attribution.sources,
"error_reason": "history_missing_checkpoint",
})),
);
let mut oversized_attribution = attribution;
oversized_attribution.sources[0].tool_name = "a".repeat(MAX_MCP_ATTRIBUTION_BYTES);
responses_metadata.mcp_attribution = Some(oversized_attribution);
@@ -603,7 +634,7 @@ fn responses_request_limits_internal_metadata_to_resolved_first_party_https_endp
.get(MCP_ATTRIBUTION_CLIENT_METADATA_KEY)
.map(String::as_str)
}),
Some(r#"{"status":"attribution_error"}"#),
Some(r#"{"status":"attribution_error","error_reason":"payload_too_large"}"#),
);
Ok(())
}
codex-rs/core/src/responses_metadata.rs
+10 -1
View File
@@ -1,5 +1,6 @@
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::io::ErrorKind;
use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
@@ -359,7 +360,15 @@ impl CodexResponsesMetadata {
}
if include_internal && let Some(attribution) = &self.mcp_attribution {
let serialized = to_json_string_bounded(attribution, MAX_MCP_ATTRIBUTION_BYTES)
.unwrap_or_else(|_| r#"{"status":"attribution_error"}"#.to_string());
.unwrap_or_else(|error| {
if error.io_error_kind() == Some(ErrorKind::WriteZero) {
r#"{"status":"attribution_error","error_reason":"payload_too_large"}"#
.to_string()
} else {
r#"{"status":"attribution_error","error_reason":"serialization_failed"}"#
.to_string()
}
});
client_metadata.insert(MCP_ATTRIBUTION_CLIENT_METADATA_KEY.to_string(), serialized);
}
client_metadata
codex-rs/core/src/session/tests.rs
+2
View File
@@ -5632,6 +5632,7 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore
.record_mcp_source(source.clone());
let expected = McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source],
};
session
@@ -6288,6 +6289,7 @@ async fn response_metadata_builders_capture_fresh_mcp_attribution() {
.await;
let expected = Some(McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source],
});
assert_eq!(before.mcp_attribution, Some(McpAttribution::default()));
codex-rs/core/src/tools/executed_tool_calls/mcp_attribution.rs
+27 -16
View File
@@ -7,6 +7,7 @@ use std::sync::MutexGuard;
use codex_history::InitialHistory;
use codex_history::RolloutItem;
use codex_protocol::mcp::McpAttribution;
use codex_protocol::mcp::McpAttributionErrorReason;
use codex_protocol::mcp::McpAttributionSource;
use codex_protocol::mcp::McpAttributionStatus;
@@ -56,25 +57,24 @@ impl McpAttributionRecorder {
for item in history.get_rollout_items() {
match item {
RolloutItem::ResponseItem(envelope) => {
if let Some(checkpoint) = envelope
.metadata
.as_ref()
.and_then(|metadata| metadata.mcp_attribution.as_ref())
if let Some(metadata) = envelope.metadata.as_ref()
&& let Some(checkpoint) = metadata.mcp_attribution.as_ref()
{
found_checkpoint = true;
state.merge_checkpoint(checkpoint);
}
}
RolloutItem::Compacted(compacted) => {
for checkpoint in compacted
for metadata in compacted
.replacement_history
.iter()
.flatten()
.filter_map(|envelope| envelope.metadata.as_ref())
.filter_map(|metadata| metadata.mcp_attribution.as_ref())
{
found_checkpoint = true;
state.merge_checkpoint(checkpoint);
if let Some(checkpoint) = metadata.mcp_attribution.as_ref() {
found_checkpoint = true;
state.merge_checkpoint(checkpoint);
}
}
}
_ => {}
@@ -82,7 +82,7 @@ impl McpAttributionRecorder {
}
if !found_checkpoint {
// Pre-attribution history cannot establish that earlier context was MCP-free.
state.mark_error();
state.mark_error(McpAttributionErrorReason::HistoryMissingCheckpoint);
}
Self(Arc::new(Mutex::new(state)))
}
@@ -90,7 +90,7 @@ impl McpAttributionRecorder {
fn lock(&self) -> MutexGuard<'_, State> {
self.0.lock().unwrap_or_else(|poisoned| {
let mut state = poisoned.into_inner();
state.mark_error();
state.mark_error(McpAttributionErrorReason::RecorderPoisoned);
state
})
}
@@ -116,20 +116,27 @@ impl McpAttributionRecorder {
}
impl State {
fn mark_error(&mut self) {
fn mark_error(&mut self, reason: McpAttributionErrorReason) {
if self.attribution.status != McpAttributionStatus::AttributionError {
self.attribution.status = McpAttributionStatus::AttributionError;
self.attribution.error_reason = Some(reason);
self.revision += 1;
}
}
fn merge_checkpoint(&mut self, checkpoint: &McpAttribution) {
if checkpoint.status == McpAttributionStatus::AttributionError
|| (checkpoint.status == McpAttributionStatus::None && !checkpoint.sources.is_empty())
if checkpoint.status == McpAttributionStatus::AttributionError {
self.mark_error(
checkpoint
.error_reason
.unwrap_or(McpAttributionErrorReason::RestoredErrorUnknown),
);
} else if (checkpoint.status == McpAttributionStatus::None
&& !checkpoint.sources.is_empty())
|| (checkpoint.status == McpAttributionStatus::Complete
&& checkpoint.sources.is_empty())
{
self.mark_error();
self.mark_error(McpAttributionErrorReason::CheckpointInvalid);
}
for source in &checkpoint.sources {
self.record(source.clone(), /*restoring*/ true);
@@ -145,7 +152,7 @@ impl State {
.find(|existing| SourceIdentity::from(*existing) == identity)
{
if restoring && existing.first_turn_id != source.first_turn_id {
self.mark_error();
self.mark_error(McpAttributionErrorReason::CheckpointSourceConflict);
}
return;
}
@@ -153,7 +160,11 @@ impl State {
|| source.tool_name.is_empty()
|| source.first_turn_id.is_empty()
{
self.mark_error();
self.mark_error(if restoring {
McpAttributionErrorReason::CheckpointInvalid
} else {
McpAttributionErrorReason::SourceInvalid
});
return;
}
if self.attribution.status == McpAttributionStatus::None {
codex-rs/core/src/tools/executed_tool_calls/mcp_attribution_tests.rs
+110
View File
@@ -44,6 +44,7 @@ fn records_the_first_turn_for_each_unique_source() {
recorder.snapshot(),
McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source("search", "turn_1"), source("fetch", "turn_2")],
}
);
@@ -53,10 +54,12 @@ fn records_the_first_turn_for_each_unique_source() {
fn restores_cumulative_item_and_compaction_checkpoints() {
let initial = McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source("search", "turn_1")],
};
let cumulative = McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source("search", "turn_1"), source("fetch", "turn_2")],
};
let history = InitialHistory::Forked(vec![
@@ -89,6 +92,7 @@ fn legacy_or_conflicting_history_is_not_complete() {
McpAttributionRecorder::new(&legacy).snapshot(),
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(McpAttributionErrorReason::HistoryMissingCheckpoint),
sources: Vec::new(),
}
);
@@ -98,6 +102,7 @@ fn legacy_or_conflicting_history_is_not_complete() {
.map(|turn_id| {
RolloutItem::ResponseItem(envelope(Some(McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source("search", turn_id)],
})))
})
@@ -107,11 +112,116 @@ fn legacy_or_conflicting_history_is_not_complete() {
McpAttributionRecorder::new(&history).snapshot(),
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(McpAttributionErrorReason::CheckpointSourceConflict),
sources: vec![source("search", "turn_1")],
}
);
}
#[test]
fn first_error_reason_survives_checkpoints_and_later_sources() {
let recorder = McpAttributionRecorder::default();
recorder.record(source("", "turn_1"));
recorder.record(source("search", "turn_2"));
let (snapshot, _) = recorder
.checkpoint(/*force*/ true)
.expect("error checkpoint");
let checkpoint = envelope(Some(snapshot));
let mut compacted: CompactedItem =
serde_json::from_value(serde_json::json!({"message": "summary"}))
.expect("compaction checkpoint");
compacted.replacement_history = Some(vec![checkpoint.clone()]);
for item in [
RolloutItem::ResponseItem(checkpoint),
RolloutItem::Compacted(compacted),
] {
let restored = McpAttributionRecorder::new(&InitialHistory::Forked(vec![item]));
restored.record(source("", "turn_3"));
assert_eq!(
restored.snapshot(),
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(McpAttributionErrorReason::SourceInvalid),
sources: vec![source("search", "turn_2")],
},
);
}
}
#[test]
fn poisoned_recorder_reports_a_bounded_reason() {
let recorder = McpAttributionRecorder::default();
let poisoned = recorder.clone();
assert!(
std::thread::spawn(move || {
let _state = poisoned.0.lock().expect("unpoisoned recorder");
panic!("poison recorder");
})
.join()
.is_err()
);
assert_eq!(
recorder.snapshot(),
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(McpAttributionErrorReason::RecorderPoisoned),
sources: Vec::new(),
}
);
}
#[test]
fn restored_diagnostics_do_not_change_attribution_state() {
let checkpoint = envelope(Some(McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: Some(McpAttributionErrorReason::SourceInvalid),
sources: vec![source("search", "turn_1")],
}));
assert_eq!(
McpAttributionRecorder::new(&InitialHistory::Forked(vec![RolloutItem::ResponseItem(
checkpoint
),]))
.snapshot(),
McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: vec![source("search", "turn_1")],
}
);
for (attribution, expected_reason) in [
(
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: None,
sources: Vec::new(),
},
McpAttributionErrorReason::RestoredErrorUnknown,
),
(
McpAttribution {
status: McpAttributionStatus::Complete,
error_reason: None,
sources: Vec::new(),
},
McpAttributionErrorReason::CheckpointInvalid,
),
] {
assert_eq!(
McpAttributionRecorder::new(&InitialHistory::Forked(vec![RolloutItem::ResponseItem(
envelope(Some(attribution))
),]))
.snapshot(),
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(expected_reason),
sources: Vec::new(),
}
);
}
}
#[test]
fn acknowledging_an_older_checkpoint_does_not_clear_newer_changes() {
let recorder = McpAttributionRecorder::default();
codex-rs/history/src/lib.rs
+1
View File
@@ -113,6 +113,7 @@ where
Ok(Some(serde_json::from_value(value).unwrap_or_else(|_| {
McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: None,
sources: Vec::new(),
}
})))
codex-rs/history/src/tests.rs
+42
View File
@@ -181,12 +181,54 @@ fn mcp_checkpoint_handles_missing_or_unknown_identity() -> Result<()> {
restored.mcp_attribution,
Some(McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: None,
sources: Vec::new(),
})
);
Ok(())
}
#[test]
fn mcp_error_diagnostics_do_not_invalidate_checkpoints() -> Result<()> {
let metadata = CodexHarnessMetadata {
mcp_attribution: Some(McpAttribution {
status: McpAttributionStatus::AttributionError,
error_reason: Some(
codex_protocol::mcp::McpAttributionErrorReason::HistoryMissingCheckpoint,
),
sources: Vec::new(),
}),
..Default::default()
};
for (reason, expected_reason) in [
(None, None),
(
Some(json!("future_reason")),
Some(codex_protocol::mcp::McpAttributionErrorReason::Unknown),
),
(Some(json!({"invalid": true})), None),
] {
let mut serialized = serde_json::to_value(&metadata)?;
let attribution = serialized["mcp_attribution"]
.as_object_mut()
.expect("attribution object");
if let Some(reason) = reason {
attribution.insert("error_reason".to_string(), reason);
} else {
attribution.remove("error_reason");
}
let restored: CodexHarnessMetadata = serde_json::from_value(serialized)?;
let mut expected = metadata.clone();
expected
.mcp_attribution
.as_mut()
.expect("attribution checkpoint")
.error_reason = expected_reason;
assert_eq!(restored, expected);
}
Ok(())
}
#[test]
/// Keeps legacy response-item rollout lines readable and byte-shape compatible.
fn response_item_rollout_line_preserves_shape() -> Result<()> {
codex-rs/protocol/src/mcp.rs
+39
View File
@@ -5,6 +5,7 @@
//! `schemars`) so they can be embedded in Codex's own protocol structures.
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Deserializer;
use serde::Serialize;
use std::collections::HashMap;
use ts_rs::TS;
@@ -62,6 +63,27 @@ pub enum McpAttributionStatus {
AttributionError,
}
/// Bounded diagnostics for an attribution error, never a policy decision.
///
/// Recorder reasons identify the first observed error and survive checkpoints.
/// Payload reasons describe a request-serialization fallback and are not checkpointed.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum McpAttributionErrorReason {
HistoryMissingCheckpoint,
CheckpointInvalid,
CheckpointSourceConflict,
SourceInvalid,
RecorderPoisoned,
/// An earlier error checkpoint did not retain its cause.
RestoredErrorUnknown,
PayloadTooLarge,
SerializationFailed,
#[serde(other)]
Unknown,
}
/// One source and the first runtime turn in which Codex recorded it.
///
/// Server and tool names are retained when no stable identifier is available;
@@ -84,10 +106,27 @@ pub struct McpAttributionSource {
#[serde(deny_unknown_fields)]
pub struct McpAttribution {
pub status: McpAttributionStatus,
/// First observed attribution error, absent for non-error states.
#[serde(
default,
skip_serializing_if = "Option::is_none",
deserialize_with = "deserialize_mcp_attribution_error_reason"
)]
pub error_reason: Option<McpAttributionErrorReason>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub sources: Vec<McpAttributionSource>,
}
fn deserialize_mcp_attribution_error_reason<'de, D>(
deserializer: D,
) -> Result<Option<McpAttributionErrorReason>, D::Error>
where
D: Deserializer<'de>,
{
let value = serde_json::Value::deserialize(deserializer)?;
Ok(serde_json::from_value(value).ok())
}
/// Bounded app-resource provenance retained across a compaction checkpoint.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct McpResourceOriginCheckpoint {