openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Replace unavailable image references in model requests (#57585)

Browse Source 
## Why

Image files can become unavailable between turns. Model requests should reflect their current availability while preserving the original references in saved history.

## What changed

- Add opt-in availability checks through `AttachmentStore::is_resolvable`, retaining references by default and on backend or access errors.
- Check distinct file IDs concurrently for each prepared request and replace confirmed-missing images with `[image unavailable]` in messages and tool outputs.
- Apply the same handling to compaction requests, with background image checks running outside the session lock without delaying foreground turn completion.

## Testing

Add unit tests and multi-turn scenarios covering placeholders, preserved history and annotations, concurrent deduplicated checks, backend errors, and nonblocking background compaction.

GitOrigin-RevId: ec98c7014c6e6b5063d87b943d2cd66a9ce7c1e3
This commit is contained in:
Krish Chainani
2026-10-27 22:59:22 +00:00
committed by copyberry
parent 5078420b01
commit cb66dbecf2
21 changed files with 876 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/attachment-store/src/lib.rs
+13 -1
View File
@@ -1,4 +1,4 @@
//! Storage-neutral attachment persistence interfaces.
//! Storage-neutral attachment persistence and availability checks.
use std::collections::HashMap;
use std::error::Error;
@@ -21,6 +21,9 @@ pub type UploadFuture<'a> =
pub type ResolveFuture<'a> =
Pin<Box<dyn Future<Output = AttachmentStoreResult<AttachmentMetadata>> + Send + 'a>>;
/// Future returned by [`AttachmentStore::is_resolvable`].
pub type IsResolvableFuture<'a> = Pin<Box<dyn Future<Output = bool> + Send + 'a>>;
/// Uploads attachments and resolves metadata for stored files.
///
/// Implementations choose whether uploaded attachments remain inline or become file references.
@@ -32,6 +35,15 @@ pub trait AttachmentStore: Send + Sync {
/// Resolves metadata and, when requested, a fresh URL for a stored attachment.
fn resolve<'a>(&'a self, request: ResolveRequest<'a>) -> ResolveFuture<'a>;
/// Checks whether a file reference should be retained in a model request.
///
/// The default retains all references, including provider-owned IDs this store cannot
/// resolve. Authoritative stores may opt in to checking and return `false` only when the
/// file is confirmed missing. Backend and access errors must retain the reference.
fn is_resolvable<'a>(&'a self, _file_id: &'a str) -> IsResolvableFuture<'a> {
Box::pin(async { true })
}
}
/// Attachment data supplied to [`AttachmentStore::upload`].
codex-rs/attachment-store/src/lib_tests.rs
+3 -1
View File
@@ -66,9 +66,11 @@ async fn inline_store_preserves_image_bytes() {
}
}
/// The inline store reports file references as missing because it never uploads them.
/// The inline store preserves opaque file references even though it cannot resolve them.
#[tokio::test]
async fn inline_store_cannot_resolve_file_references() {
assert!(InlineAttachmentStore.is_resolvable("file_123").await);
let error = InlineAttachmentStore
.resolve(ResolveRequest {
file_id: "file_123",
codex-rs/core/src/agent/control_tests.rs
+5 -1
View File
@@ -489,7 +489,11 @@ async fn mcp_attribution_in_constructed_request(thread: &CodexThread) -> McpAttr
.session
.clone_history()
.await
.for_prompt(&step_context.settings.model_info.input_modalities),
.for_prompt(
&step_context.settings.model_info.input_modalities,
thread.session.services.image_store.as_ref(),
)
.await,
&step_context,
thread.session.get_prompt_base_instructions().await,
thread
codex-rs/core/src/compact.rs
+5 -1
View File
@@ -299,7 +299,11 @@ async fn run_compact_task_inner_impl(
// Clone is required because of the loop
let mut turn_input = history
.clone()
.for_prompt(&turn_context.model_info().input_modalities);
.for_prompt(
&turn_context.model_info().input_modalities,
sess.services.image_store.as_ref(),
)
.await;
sess.services
.executed_tool_calls
.attach_to_compaction_prompt(&mut turn_input);
codex-rs/core/src/compact_remote_v2_attempt.rs
+21 -4
View File
@@ -14,6 +14,7 @@ use crate::compact::CompactionAnalyticsDetails;
use crate::compact_remote_history::trim_function_call_history_to_fit_context_window;
use crate::context::world_state::WorldState;
use crate::context_manager::ContextManager;
use crate::context_manager::replace_unresolvable_image_files;
use crate::responses_metadata::CompactionTurnMetadata;
use crate::session::CompactionPrefix;
use crate::session::session::Session;
@@ -23,12 +24,14 @@ use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionTrigger;
use codex_history::CodexHarnessMetadata;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::TokenUsage;
use codex_rollout_trace::CompactionTraceContext;
use tokio_util::sync::CancellationToken;
use tracing::info;
pub(super) struct RemoteCompactV2Attempt {
@@ -129,7 +132,11 @@ pub(super) async fn run_remote_compact_v2_attempt(
.is_enabled()
.then(|| history.raw_items().cloned().collect());
let (mut input, metadata): (Vec<_>, Vec<_>) = history
.for_prompt_annotated(&model_info.input_modalities)
.for_prompt_annotated(
&model_info.input_modalities,
sess.services.image_store.as_ref(),
)
.await
.into_iter()
.map(|envelope| (envelope.item, envelope.metadata))
.unzip();
@@ -232,7 +239,8 @@ pub(super) async fn run_remote_compact_v2_attempt(
})
}
pub(crate) fn capture_background_input(
// Defer normalization and attribution until the caller releases the session lock.
pub(crate) async fn capture_background_input(
sess: &Session,
step_context: &StepContext,
mut snapshot: ContextManager,
@@ -264,8 +272,9 @@ pub(crate) fn capture_background_input(
step_context.settings.model_info.usable_context_window(),
&base_instructions,
);
snapshot.normalize_history(&step_context.settings.model_info.input_modalities);
let (mut input, metadata): (Vec<_>, Vec<_>) = snapshot
.for_prompt_annotated(&step_context.settings.model_info.input_modalities)
.into_annotated_items()
.into_iter()
.map(|envelope| (envelope.item, envelope.metadata))
.unzip();
@@ -287,8 +296,16 @@ pub(crate) async fn run_background_compaction(
step_context: Arc<StepContext>,
cutoff_world_state: Arc<WorldState>,
phase: CompactionPhase,
prepared_input: BackgroundCompactionInput,
mut prepared_input: BackgroundCompactionInput,
shutdown_cancellation: &CancellationToken,
) -> CodexResult<BackgroundCompactionCandidate> {
tokio::select! {
_ = shutdown_cancellation.cancelled() => return Err(CodexErr::TurnAborted),
() = replace_unresolvable_image_files(
prepared_input.input.iter_mut(),
sess.services.image_store.as_ref(),
) => {},
}
let prefix = prepared_input.prefix;
let metadata = CompactionTurnMetadata::new(
CompactionTrigger::Auto,
codex-rs/core/src/context_manager/history.rs
+21 -4
View File
@@ -8,6 +8,7 @@
//! Token estimates charge item content rather than transport metadata.
//! Oversized instructions keep an incomplete excerpt for bounded root review, including
//! sources recovered from legacy Guardian checkpoints before their raw history is dropped.
//! Prompt projections replace unavailable file images without changing retained history.
#[path = "history_user_authorization.rs"]
mod user_authorization;
@@ -21,6 +22,7 @@ use crate::context::world_state::WorldState;
use crate::context::world_state::WorldStateSnapshot;
use crate::context::world_state::WorldStateUpdate;
use crate::context_manager::normalize;
use crate::context_manager::replace_unresolvable_image_files;
use crate::event_mapping::has_non_contextual_dev_message_content;
use crate::event_mapping::is_contextual_dev_message_content;
use crate::event_mapping::is_contextual_user_message_content;
@@ -31,6 +33,7 @@ use crate::session::turn_context::TurnContext;
use crate::utils::json::serialized_json_bytes;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_attachment_store::AttachmentStore;
use codex_context_fragments::set_annotated_content;
use codex_context_fragments::to_annotated_content;
use codex_extension_api::ConversationHistorySnapshot;
@@ -600,19 +603,33 @@ impl ContextManager {
/// Returns the history prepared for sending to the model. This applies a proper
/// normalization and drops un-suited items. Unsupported image and audio content
/// is stripped from messages and tool outputs according to `input_modalities`.
pub(crate) fn for_prompt(self, input_modalities: &[InputModality]) -> Vec<ResponseItem> {
self.for_prompt_annotated(input_modalities)
/// Unavailable file images are replaced with text placeholders.
pub(crate) async fn for_prompt(
self,
input_modalities: &[InputModality],
image_store: &dyn AttachmentStore,
) -> Vec<ResponseItem> {
self.for_prompt_annotated(input_modalities, image_store)
.await
.into_iter()
.map(ResponseItemEnvelope::into_item)
.collect()
}
/// Returns normalized history envelopes for internal consumers that must retain metadata.
pub(crate) fn for_prompt_annotated(
pub(crate) async fn for_prompt_annotated(
mut self,
input_modalities: &[InputModality],
image_store: &dyn AttachmentStore,
) -> Vec<ResponseItemEnvelope> {
self.normalize_history(input_modalities);
replace_unresolvable_image_files(
Arc::make_mut(&mut self.items)
.iter_mut()
.map(|envelope| &mut envelope.item),
image_store,
)
.await;
Arc::unwrap_or_clone(self.items)
}
@@ -965,7 +982,7 @@ impl ContextManager {
/// 1. every call (function/custom) has a corresponding output entry
/// 2. every output has a corresponding call entry or names an external tool event
/// 3. unsupported image and audio content is stripped from messages and tool outputs
fn normalize_history(&mut self, input_modalities: &[InputModality]) {
pub(crate) fn normalize_history(&mut self, input_modalities: &[InputModality]) {
let items = Arc::make_mut(&mut self.items);
// all function/tool calls must have a corresponding output
codex-rs/core/src/context_manager/history_tests.rs
+206 -50
View File
@@ -5,8 +5,10 @@ use crate::context::world_state::SectionTransition;
use crate::context::world_state::WorldState;
use crate::context::world_state::WorldStateSection;
use crate::context::world_state::WorldStateUpdateContent;
use crate::image_preparation::tests::FailingAttachmentStore;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_attachment_store::InlineAttachmentStore;
use codex_history::CodexHarnessMetadata;
use codex_history::ResponseItemEnvelope;
use codex_protocol::AgentPath;
@@ -973,13 +975,18 @@ fn inter_agent_assistant_messages_are_turn_boundaries() {
assert!(is_user_turn_boundary(&item));
}
#[test]
fn for_prompt_preserves_inter_agent_assistant_messages() {
#[tokio::test]
async fn for_prompt_preserves_inter_agent_assistant_messages() {
let item = inter_agent_assistant_msg("continue");
let history = create_history_with_items(vec![item.clone()]);
assert_eq!(raw_items(&history), std::slice::from_ref(&item));
assert_eq!(history.for_prompt(&default_input_modalities()), vec![item]);
assert_eq!(
history
.for_prompt(&default_input_modalities(), &InlineAttachmentStore)
.await,
vec![item]
);
}
#[test]
@@ -1079,8 +1086,8 @@ fn record_annotated_items_preserves_metadata_while_processing_item(
);
}
#[test]
fn for_prompt_annotated_preserves_metadata_while_normalizing_item() {
#[tokio::test]
async fn for_prompt_annotated_preserves_metadata_while_normalizing_item() {
let envelope = ResponseItemEnvelope {
item: ResponseItem::Message {
status: None,
@@ -1106,7 +1113,9 @@ fn for_prompt_annotated_preserves_metadata_while_normalizing_item() {
let mut history = ContextManager::new();
history.replace_annotated(vec![envelope.clone()]);
let normalized = history.for_prompt_annotated(&[InputModality::Text]);
let normalized = history
.for_prompt_annotated(&[InputModality::Text], &InlineAttachmentStore)
.await;
assert_eq!(normalized.len(), 1);
assert_eq!(normalized[0].metadata, envelope.metadata);
@@ -1164,8 +1173,101 @@ fn total_token_usage_includes_all_items_after_last_model_generated_item() {
);
}
#[test]
fn for_prompt_strips_media_when_model_does_not_support_it() {
/// Unavailable images become harness-attributed placeholders without changing saved history or
/// neighboring inline content and its annotations.
#[tokio::test]
async fn for_prompt_replaces_unavailable_message_images() {
let original: ResponseItem = from_value(json!({
"type": "message", "role": "user",
"content": [
{"type": "input_text", "text": "look"},
{"type": "input_image", "file_id": "expired"},
{"type": "input_image", "image_url": "data:image/png;base64,aW1hZ2U="}
],
"internal_chat_message_metadata_passthrough": {
"content_item_kinds": ["user.text", "user.image", "user.image"]
}
}))
.unwrap();
let history = create_history_with_items(vec![original.clone()]);
let projected = history
.clone()
.for_prompt_annotated(
&[
InputModality::Text,
InputModality::Image,
InputModality::Audio,
],
&FailingAttachmentStore,
)
.await;
let expected: ResponseItem = from_value(json!({
"type": "message", "role": "user",
"content": [
{"type": "input_text", "text": "look"},
{"type": "input_text", "text": "[image unavailable]"},
{"type": "input_image", "image_url": "data:image/png;base64,aW1hZ2U="}
],
"internal_chat_message_metadata_passthrough": {
"content_item_kinds": ["user.text", "images.unavailable", "user.image"],
"content_item_metadata": [{}, {"harness_injected": true}, {}]
}
}))
.unwrap();
assert_eq!(projected, vec![ResponseItemEnvelope::new(expected)]);
assert_eq!(raw_items(&history), vec![original]);
}
/// Both tool output formats retain inline images and text when a neighboring file expires.
#[tokio::test]
async fn for_prompt_replaces_unavailable_tool_images() {
for (call, output) in [
(
json!({"type": "function_call", "call_id": "call-1", "name": "view_image", "arguments": "{}"}),
json!({"type": "function_call_output", "call_id": "call-1"}),
),
(
json!({"type": "custom_tool_call", "call_id": "call-1", "name": "js_repl", "input": "view_image"}),
json!({"type": "custom_tool_call_output", "call_id": "call-1"}),
),
] {
let call: ResponseItem = from_value(call).unwrap();
let mut original = output.clone();
original["output"] = json!([
{"type": "input_text", "text": "result"},
{"type": "input_image", "file_id": "expired"},
{"type": "input_image", "image_url": "data:image/png;base64,aW1hZ2U="}
]);
let original: ResponseItem = from_value(original).unwrap();
let history = create_history_with_items(vec![call.clone(), original.clone()]);
let mut expected = output;
expected["output"] = json!([
{"type": "input_text", "text": "result"},
{"type": "input_text", "text": "[image unavailable]"},
{"type": "input_image", "image_url": "data:image/png;base64,aW1hZ2U="}
]);
let expected: ResponseItem = from_value(expected).unwrap();
assert_eq!(
history
.clone()
.for_prompt(
&[
InputModality::Text,
InputModality::Image,
InputModality::Audio
],
&FailingAttachmentStore,
)
.await,
vec![call.clone(), expected]
);
assert_eq!(raw_items(&history), vec![call, original]);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn for_prompt_strips_media_when_model_does_not_support_it() {
let items = vec![
ResponseItem::Message {
status: None,
@@ -1268,7 +1370,9 @@ fn for_prompt_strips_media_when_model_does_not_support_it() {
let fully_supported_items = items.clone();
let history = create_history_with_items(items);
let text_only_modalities = vec![InputModality::Text];
let stripped = history.for_prompt(&text_only_modalities);
let stripped = history
.for_prompt(&text_only_modalities, &InlineAttachmentStore)
.await;
let expected = vec![
ResponseItem::Message {
@@ -1372,11 +1476,16 @@ fn for_prompt_strips_media_when_model_does_not_support_it() {
];
assert_eq!(stripped, expected);
assert_eq!(
create_history_with_items(fully_supported_items.clone()).for_prompt(&[
InputModality::Text,
InputModality::Image,
InputModality::Audio,
]),
create_history_with_items(fully_supported_items.clone())
.for_prompt(
&[
InputModality::Text,
InputModality::Image,
InputModality::Audio,
],
&InlineAttachmentStore
)
.await,
fully_supported_items
);
@@ -1401,7 +1510,9 @@ fn for_prompt_strips_media_when_model_does_not_support_it() {
phase: None,
internal_chat_message_metadata_passthrough: None,
}]);
let preserved = with_images.for_prompt(&modalities);
let preserved = with_images
.for_prompt(&modalities, &InlineAttachmentStore)
.await;
assert_eq!(preserved.len(), 1);
if let ResponseItem::Message { content, .. } = &preserved[0] {
assert_eq!(content.len(), 2);
@@ -1423,13 +1534,18 @@ fn for_prompt_strips_media_when_model_does_not_support_it() {
};
let with_audio = create_history_with_items(vec![audio_message.clone()]);
assert_eq!(
with_audio.for_prompt(&[InputModality::Text, InputModality::Audio]),
with_audio
.for_prompt(
&[InputModality::Text, InputModality::Audio],
&InlineAttachmentStore
)
.await,
vec![audio_message]
);
}
#[test]
fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
#[tokio::test]
async fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
let history = create_history_with_items(vec![
ResponseItem::ImageGenerationCall {
encrypted_content: None,
@@ -1453,7 +1569,9 @@ fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
]);
assert_eq!(
history.for_prompt(&default_input_modalities()),
history
.for_prompt(&default_input_modalities(), &InlineAttachmentStore)
.await,
vec![
ResponseItem::ImageGenerationCall {
encrypted_content: None,
@@ -1478,8 +1596,8 @@ fn for_prompt_preserves_image_generation_calls_when_images_are_supported() {
);
}
#[test]
fn for_prompt_clears_image_generation_result_when_images_are_unsupported() {
#[tokio::test]
async fn for_prompt_clears_image_generation_result_when_images_are_unsupported() {
let history = create_history_with_items(vec![
ResponseItem::Message {
status: None,
@@ -1503,7 +1621,9 @@ fn for_prompt_clears_image_generation_result_when_images_are_unsupported() {
]);
assert_eq!(
history.for_prompt(&[InputModality::Text]),
history
.for_prompt(&[InputModality::Text], &InlineAttachmentStore)
.await,
vec![
ResponseItem::Message {
encrypted_content: None,
@@ -1657,8 +1777,8 @@ fn remove_first_item_handles_local_shell_pair() {
assert_eq!(raw_items(&h), vec![]);
}
#[test]
fn drop_last_n_user_turns_preserves_prefix() {
#[tokio::test]
async fn drop_last_n_user_turns_preserves_prefix() {
let items = vec![
assistant_msg("session prefix item"),
user_msg("u1"),
@@ -1671,7 +1791,9 @@ fn drop_last_n_user_turns_preserves_prefix() {
let mut history = create_history_with_items(items);
history.drop_last_n_user_turns(/*num_turns*/ 1);
assert_eq!(
history.for_prompt(&modalities),
history
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
vec![
assistant_msg("session prefix item"),
user_msg("u1"),
@@ -1688,7 +1810,10 @@ fn drop_last_n_user_turns_preserves_prefix() {
]);
history.drop_last_n_user_turns(/*num_turns*/ 99);
assert_eq!(
history.clone().for_prompt(&modalities),
history
.clone()
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
vec![assistant_msg("session prefix item")]
);
// With no remaining instruction boundary, rollback must not revoke facts from a
@@ -1792,8 +1917,8 @@ fn rollback_removes_assistant_sources_recorded_ahead_of_queued_input() {
);
}
#[test]
fn drop_last_n_user_turns_ignores_session_prefix_user_messages() {
#[tokio::test]
async fn drop_last_n_user_turns_ignores_session_prefix_user_messages() {
let items = vec![
user_input_text_msg("<environment_context>ctx</environment_context>"),
user_input_text_msg(
@@ -1833,7 +1958,9 @@ fn drop_last_n_user_turns_ignores_session_prefix_user_messages() {
];
assert_eq!(
history.for_prompt(&modalities),
history
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
expected_prefix_and_first_turn
);
@@ -1869,7 +1996,12 @@ fn drop_last_n_user_turns_ignores_session_prefix_user_messages() {
assistant_msg("turn 2 assistant"),
]);
history.drop_last_n_user_turns(/*num_turns*/ 2);
assert_eq!(history.for_prompt(&modalities), expected_prefix_only);
assert_eq!(
history
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
expected_prefix_only
);
let mut history = create_history_with_items(vec![
user_input_text_msg("<environment_context>ctx</environment_context>"),
@@ -1889,11 +2021,16 @@ fn drop_last_n_user_turns_ignores_session_prefix_user_messages() {
assistant_msg("turn 2 assistant"),
]);
history.drop_last_n_user_turns(/*num_turns*/ 3);
assert_eq!(history.for_prompt(&modalities), expected_prefix_only);
assert_eq!(
history
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
expected_prefix_only
);
}
#[test]
fn drop_last_n_user_turns_trims_context_updates_above_rolled_back_turn() {
#[tokio::test]
async fn drop_last_n_user_turns_trims_context_updates_above_rolled_back_turn() {
let items = vec![
assistant_msg("session prefix item"),
user_input_text_msg("turn 1 user"),
@@ -1927,7 +2064,10 @@ fn drop_last_n_user_turns_trims_context_updates_above_rolled_back_turn() {
history.drop_last_n_user_turns(/*num_turns*/ 1);
assert_eq!(
history.clone().for_prompt(&modalities),
history
.clone()
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
vec![
assistant_msg("session prefix item"),
user_input_text_msg("turn 1 user"),
@@ -2014,8 +2154,8 @@ fn drop_last_n_user_turns_preserves_annotations_for_surviving_developer_fragment
);
}
#[test]
fn drop_last_n_user_turns_trims_saved_prefix_update_above_rolled_back_turn() {
#[tokio::test]
async fn drop_last_n_user_turns_trims_saved_prefix_update_above_rolled_back_turn() {
let items = vec![
assistant_msg("session prefix item"),
user_input_text_msg("turn 1 user"),
@@ -2032,7 +2172,9 @@ fn drop_last_n_user_turns_trims_saved_prefix_update_above_rolled_back_turn() {
history.drop_last_n_user_turns(/*num_turns*/ 1);
assert_eq!(
history.for_prompt(&modalities),
history
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
vec![
assistant_msg("session prefix item"),
user_input_text_msg("turn 1 user"),
@@ -2041,8 +2183,8 @@ fn drop_last_n_user_turns_trims_saved_prefix_update_above_rolled_back_turn() {
);
}
#[test]
fn drop_last_n_user_turns_clears_reference_context_for_mixed_developer_context_bundles() {
#[tokio::test]
async fn drop_last_n_user_turns_clears_reference_context_for_mixed_developer_context_bundles() {
let items = vec![
user_input_text_msg("turn 1 user"),
assistant_msg("turn 1 assistant"),
@@ -2063,7 +2205,10 @@ fn drop_last_n_user_turns_clears_reference_context_for_mixed_developer_context_b
history.drop_last_n_user_turns(/*num_turns*/ 1);
assert_eq!(
history.clone().for_prompt(&modalities),
history
.clone()
.for_prompt(&modalities, &InlineAttachmentStore)
.await,
vec![
user_input_text_msg("turn 1 user"),
assistant_msg("turn 1 assistant"),
@@ -2098,8 +2243,8 @@ fn remove_first_item_handles_custom_tool_pair() {
assert_eq!(raw_items(&h), vec![]);
}
#[test]
fn normalization_retains_local_shell_outputs() {
#[tokio::test]
async fn normalization_retains_local_shell_outputs() {
let items = vec![
ResponseItem::LocalShellCall {
encrypted_content: None,
@@ -2127,7 +2272,9 @@ fn normalization_retains_local_shell_outputs() {
let modalities = default_input_modalities();
let history = create_history_with_items(items.clone());
let normalized = history.for_prompt(&modalities);
let normalized = history
.for_prompt(&modalities, &InlineAttachmentStore)
.await;
assert_eq!(normalized, items);
}
@@ -2630,8 +2777,8 @@ fn normalize_adds_missing_output_for_function_call_inserts_output() {
);
}
#[test]
fn normalize_preserves_named_function_call_output_without_call_id() {
#[tokio::test]
async fn normalize_preserves_named_function_call_output_without_call_id() {
let item = ResponseItem::FunctionCallOutput {
id: None,
call_id: None,
@@ -2642,11 +2789,16 @@ fn normalize_preserves_named_function_call_output_without_call_id() {
};
let history = create_history_with_items(vec![item.clone()]);
assert_eq!(history.for_prompt(&default_input_modalities()), vec![item]);
assert_eq!(
history
.for_prompt(&default_input_modalities(), &InlineAttachmentStore)
.await,
vec![item]
);
}
#[test]
fn for_prompt_assigns_stable_id_to_synthetic_output_without_reordering_history() {
#[tokio::test]
async fn for_prompt_assigns_stable_id_to_synthetic_output_without_reordering_history() {
let items = vec![
ResponseItem::FunctionCall {
status: None,
@@ -2672,8 +2824,12 @@ fn for_prompt_assigns_stable_id_to_synthetic_output_without_reordering_history()
},
];
let first = create_history_with_items(items.clone()).for_prompt(&default_input_modalities());
let second = create_history_with_items(items).for_prompt(&default_input_modalities());
let first = create_history_with_items(items.clone())
.for_prompt(&default_input_modalities(), &InlineAttachmentStore)
.await;
let second = create_history_with_items(items)
.for_prompt(&default_input_modalities(), &InlineAttachmentStore)
.await;
assert_eq!(
first, second,
codex-rs/core/src/context_manager/image_availability.rs
+118
View File
@@ -0,0 +1,118 @@
//! Projects unavailable file images into placeholders without changing tool attribution.
use std::collections::HashSet;
use codex_attachment_store::AttachmentStore;
use codex_context_fragments::AnnotatedContent;
use codex_context_fragments::set_annotated_content;
use codex_context_fragments::to_annotated_content;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::ContentItemMetadata;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::ImageReference;
use codex_protocol::models::ResponseItem;
use futures::future::join_all;
/// Checks distinct file IDs concurrently and replaces only unavailable image content.
pub(crate) async fn replace_unresolvable_image_files<'a>(
items: impl Iterator<Item = &'a mut ResponseItem>,
image_store: &dyn AttachmentStore,
) {
let items: Vec<_> = items.collect();
let mut file_ids = HashSet::new();
for item in &items {
match item {
ResponseItem::Message { content, .. } => {
for part in content {
if let ContentItem::InputImage {
image: ImageReference::File { file_id },
..
} = part
{
file_ids.insert(file_id.clone());
}
}
}
ResponseItem::FunctionCallOutput { output, .. }
| ResponseItem::CustomToolCallOutput { output, .. } => {
if let Some(content) = output.content_items() {
for part in content {
if let FunctionCallOutputContentItem::InputImage {
image: ImageReference::File { file_id },
..
} = part
{
file_ids.insert(file_id.clone());
}
}
}
}
_ => {}
}
}
let results = join_all(file_ids.into_iter().map(|file_id| async move {
let resolvable = image_store.is_resolvable(&file_id).await;
(file_id, resolvable)
}))
.await;
let mut unavailable_file_ids = HashSet::new();
for (file_id, resolvable) in results {
if !resolvable {
unavailable_file_ids.insert(file_id);
}
}
if unavailable_file_ids.is_empty() {
return;
}
for item in items {
match item {
ResponseItem::Message { content, .. } => {
let mut unavailable = Vec::new();
for (index, part) in content.iter().enumerate() {
if let ContentItem::InputImage {
image: ImageReference::File { file_id },
..
} = part
&& unavailable_file_ids.contains(file_id)
{
unavailable.push(index);
}
}
if unavailable.is_empty() {
continue;
}
let Some(mut content) = to_annotated_content(item) else {
continue;
};
for index in unavailable {
content[index] = AnnotatedContent::text(
"[image unavailable]",
ContentItemKind("images.unavailable".to_string()),
ContentItemMetadata::harness(),
);
}
let _ = set_annotated_content(item, content);
}
ResponseItem::FunctionCallOutput { output, .. }
| ResponseItem::CustomToolCallOutput { output, .. } => {
if let Some(content) = output.content_items_mut() {
for part in content {
if let FunctionCallOutputContentItem::InputImage {
image: ImageReference::File { file_id },
..
} = part
&& unavailable_file_ids.contains(file_id)
{
*part = FunctionCallOutputContentItem::InputText {
text: "[image unavailable]".to_string(),
};
}
}
}
}
_ => {}
}
}
}
codex-rs/core/src/context_manager/mod.rs
+2
View File
@@ -1,4 +1,5 @@
mod history;
mod image_availability;
mod normalize;
pub(crate) mod updates;
@@ -7,3 +8,4 @@ pub(crate) use history::HistoryReplacement;
pub(crate) use history::estimate_image_reference_bytes;
pub(crate) use history::estimate_item_token_count;
pub(crate) use history::is_user_turn_boundary;
pub(crate) use image_availability::replace_unresolvable_image_files;
codex-rs/core/src/guardian/input_budget.rs
+11 -2
View File
@@ -88,7 +88,11 @@ pub(crate) async fn check_pending(session: &Session, turn: &TurnContext) -> Code
let history = session
.clone_history()
.await
.for_prompt_annotated(&turn.model_info().input_modalities);
.for_prompt_annotated(
&turn.model_info().input_modalities,
session.services.image_store.as_ref(),
)
.await;
context.retain_new_instructions(&history);
context
.retain_new_native_messages(&history)
@@ -151,7 +155,12 @@ pub(crate) async fn finalize(
));
}
let history_version = history.history_version();
let history = history.for_prompt_annotated(&model.input_modalities);
let history = history
.for_prompt_annotated(
&model.input_modalities,
session.services.image_store.as_ref(),
)
.await;
// Use the history that will actually reach the model. Recompute after every
// compaction retry; evidence removed by compaction must be delivered again.
context.retain_new_instructions(&history);
codex-rs/core/src/guardian/input_budget_tests.rs
+5 -1
View File
@@ -340,7 +340,11 @@ async fn compacted_review_restores_originals_once_and_persists_the_request_prefi
let persisted = session
.clone_history()
.await
.for_prompt(&step.settings.model_info.input_modalities);
.for_prompt(
&step.settings.model_info.input_modalities,
session.services.image_store.as_ref(),
)
.await;
assert_eq!(prompt.input, persisted);
assert!(
persisted
codex-rs/core/src/guardian/request_budget.rs
+8 -1
View File
@@ -79,7 +79,14 @@ pub(crate) async fn prepare_prompt(
/*include_internal*/ true,
)?;
let mut context = retained.context.clone();
context.retain_new_instructions(&history.for_prompt_annotated(&model.input_modalities));
context.retain_new_instructions(
&history
.for_prompt_annotated(
&model.input_modalities,
session.services.image_store.as_ref(),
)
.await,
);
let context = context
.enforce_budget(
RequestBudget {
codex-rs/core/src/image_preparation.rs
+1 -1
View File
@@ -440,4 +440,4 @@ pub(crate) fn resize_image(
#[cfg(test)]
#[path = "image_preparation_tests.rs"]
mod tests;
pub(crate) mod tests;
codex-rs/core/src/image_preparation_tests.rs
+6 -1
View File
@@ -5,6 +5,7 @@ use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_attachment_store::AttachmentStoreError;
use codex_attachment_store::AttachmentStoreErrorKind;
use codex_attachment_store::InlineAttachmentStore;
use codex_attachment_store::IsResolvableFuture;
use codex_attachment_store::ResolveFuture;
use codex_attachment_store::ResolveRequest;
use codex_attachment_store::UploadFuture;
@@ -25,9 +26,13 @@ use serde_json::json;
use super::*;
struct FailingAttachmentStore;
pub(crate) struct FailingAttachmentStore;
impl AttachmentStore for FailingAttachmentStore {
fn is_resolvable<'a>(&'a self, _file_id: &'a str) -> IsResolvableFuture<'a> {
Box::pin(async { false })
}
fn upload(&self, _request: UploadRequest) -> UploadFuture<'_> {
Box::pin(async {
Err(AttachmentStoreError::new(
codex-rs/core/src/prompt_debug.rs
+5 -1
View File
@@ -106,7 +106,11 @@ pub(crate) async fn build_prompt_input_from_session(
let prompt_input = sess
.clone_history()
.await
.for_prompt(&step_context.settings.model_info.input_modalities);
.for_prompt(
&step_context.settings.model_info.input_modalities,
sess.services.image_store.as_ref(),
)
.await;
let base_instructions = sess.get_base_instructions().await;
let prompt = build_prompt(
prompt_input,
codex-rs/core/src/session/async_compaction.rs
+45 -35
View File
@@ -1,5 +1,6 @@
//! Holds the current speculative request, polled at checkpoints or awaited when compaction is due.
//! Discarded requests still charge completed usage; session shutdown cancels remaining network work.
//! Snapshot capture and launch registration hold session state; prompt preparation does not.
use std::sync::Arc;
@@ -81,42 +82,52 @@ pub(super) async fn maybe_start(
}
let base_instructions = sess.get_prompt_base_instructions().await;
let mut state = sess.state.lock().await;
let model_info = &step_context.settings.model_info;
let attempt = CompactionAttempt {
history_version: state.history.history_version(),
model_slug: model_info.slug.clone(),
compaction_hash: model_info.comp_hash.clone(),
auth_owner_generation,
};
// Stop blocks new launches; already-running compaction remains session-owned.
if turn_cancellation.is_cancelled()
|| state.shutting_down
|| state.async_compaction.last_attempt.as_ref() == Some(&attempt)
|| state.history.annotated_items().is_empty()
|| sess.services.model_client.auth_owner_generation() != auth_owner_generation
{
return;
}
let (prepared_input, tasks, shutdown_cancellation, launch, send) = {
let mut state = sess.state.lock().await;
let model_info = &step_context.settings.model_info;
let attempt = CompactionAttempt {
history_version: state.history.history_version(),
model_slug: model_info.slug.clone(),
compaction_hash: model_info.comp_hash.clone(),
auth_owner_generation,
};
// Stop blocks new launches; already-running compaction remains session-owned.
if turn_cancellation.is_cancelled()
|| state.shutting_down
|| state.async_compaction.last_attempt.as_ref() == Some(&attempt)
|| state.history.annotated_items().is_empty()
|| sess.services.model_client.auth_owner_generation() != auth_owner_generation
{
return;
}
// Capture and register without yielding, so shutdown or history replacement cannot
// overtake this request. Tool attribution still belongs to the foreground's current step.
let prepared_input = capture_background_input(
sess,
step_context,
state.history.clone(),
base_instructions,
turn_state,
auth_owner_generation,
sess.current_window_from_state(&state),
);
// Capture the arguments now; the async preparation body runs after the lock is released.
let prepared_input = capture_background_input(
sess,
step_context,
state.history.clone(),
base_instructions,
turn_state,
auth_owner_generation,
sess.current_window_from_state(&state),
);
let tasks = state.async_compaction.tasks.clone();
let shutdown_cancellation = state.async_compaction.shutdown_cancellation.clone();
// Register before unlocking so shutdown cannot finish between capture and spawning.
let launch = tasks.token();
state.async_compaction.last_attempt = Some(attempt);
let (send, result) = oneshot::channel();
state.async_compaction.result = Some(result);
(prepared_input, tasks, shutdown_cancellation, launch, send)
};
// Prepare outside the lock on the foreground task so its step cannot advance before tool
// attribution is captured. Keep normalization before attribution capture.
let prepared_input = prepared_input.await;
let task_session = Arc::clone(sess);
let task_step = Arc::clone(step_context);
let cutoff_world_state = Arc::clone(world_state);
let storage_originator = AuthStorageOriginator::from_client_name(&turn_context.originator);
state.async_compaction.last_attempt = Some(attempt);
let (send, result) = oneshot::channel();
state.async_compaction.result = Some(result);
let task = async move {
match run_background_compaction(
&task_session,
@@ -124,6 +135,7 @@ pub(super) async fn maybe_start(
cutoff_world_state,
phase,
prepared_input,
&shutdown_cancellation,
)
.await
{
@@ -135,10 +147,8 @@ pub(super) async fn maybe_start(
}
}
};
state
.async_compaction
.tasks
.spawn(storage_originator.scope(task));
tasks.spawn(storage_originator.scope(task));
drop(launch);
}
pub(super) async fn take_ready(
codex-rs/core/src/session/startup_prewarm.rs
+5 -1
View File
@@ -404,7 +404,11 @@ async fn schedule_startup_prewarm_inner(
let mut history = session
.clone_history()
.await
.for_prompt(&step_context.settings.model_info.input_modalities);
.for_prompt(
&step_context.settings.model_info.input_modalities,
session.services.image_store.as_ref(),
)
.await;
session
.services
.executed_tool_calls
codex-rs/core/src/session/turn.rs
+10 -2
View File
@@ -572,7 +572,11 @@ pub(crate) async fn run_turn(
let sampling_request_input: Vec<ResponseItem> = async {
sess.clone_history()
.await
.for_prompt(&step_context.settings.model_info.input_modalities)
.for_prompt(
&step_context.settings.model_info.input_modalities,
sess.services.image_store.as_ref(),
)
.await
}
.instrument(trace_span!("run_turn.prepare_sampling_request_input"))
.await;
@@ -1783,7 +1787,11 @@ async fn run_sampling_request(
} else {
sess.clone_history()
.await
.for_prompt(&step_context.settings.model_info.input_modalities)
.for_prompt(
&step_context.settings.model_info.input_modalities,
sess.services.image_store.as_ref(),
)
.await
};
let mut prompt_input = prompt_input;
sess.services
codex-rs/core/tests/suite/image_rollout.rs
+42 -5
View File
@@ -1,9 +1,11 @@
use anyhow::Context;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_attachment_store::AttachmentMetadata;
use codex_attachment_store::AttachmentStore;
use codex_attachment_store::AttachmentStoreError;
use codex_attachment_store::AttachmentStoreErrorKind;
use codex_attachment_store::IsResolvableFuture;
use codex_attachment_store::ResolveFuture;
use codex_attachment_store::ResolveRequest;
use codex_attachment_store::UploadFuture;
@@ -53,6 +55,7 @@ use image::ImageBuffer;
use image::Rgba;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::HashMap;
use std::fs;
use std::path::Path;
use std::sync::Arc;
@@ -64,9 +67,28 @@ const UPLOADED_FILE_ID: &str = "file_uploaded_image";
#[derive(Default)]
pub(super) struct RecordingFileAttachmentStore {
pub(super) uploads: Mutex<Vec<UploadRequest>>,
pub(super) resolve_errors: Mutex<HashMap<String, AttachmentStoreErrorKind>>,
pub(super) resolves: Mutex<HashMap<String, usize>>,
pub(super) resolve_barrier: Option<Arc<tokio::sync::Barrier>>,
}
impl AttachmentStore for RecordingFileAttachmentStore {
fn is_resolvable<'a>(&'a self, file_id: &'a str) -> IsResolvableFuture<'a> {
Box::pin(async move {
match self
.resolve(ResolveRequest {
file_id,
tags: HashMap::new(),
download_url_ttl: None,
})
.await
{
Ok(_) => true,
Err(error) => error.kind() != AttachmentStoreErrorKind::NotFound,
}
})
}
fn upload(&self, request: UploadRequest) -> UploadFuture<'_> {
self.uploads.lock().expect("uploads lock").push(request);
Box::pin(async {
@@ -77,12 +99,27 @@ impl AttachmentStore for RecordingFileAttachmentStore {
}
fn resolve<'a>(&'a self, request: ResolveRequest<'a>) -> ResolveFuture<'a> {
let file_id = request.file_id;
*self
.resolves
.lock()
.expect("resolves lock")
.entry(request.file_id.to_string())
.or_default() += 1;
let error_kind = self
.resolve_errors
.lock()
.expect("resolve errors lock")
.get(request.file_id)
.copied();
let barrier = self.resolve_barrier.clone();
Box::pin(async move {
Err(AttachmentStoreError::new(
AttachmentStoreErrorKind::NotFound,
format!("attachment `{file_id}` was not found"),
))
if let Some(barrier) = barrier {
barrier.wait().await;
}
match error_kind {
Some(kind) => Err(AttachmentStoreError::new(kind, "attachment unavailable")),
None => Ok(AttachmentMetadata::default()),
}
})
}
}
codex-rs/core/tests/suite/scenarios.rs
+262
View File
@@ -1,6 +1,8 @@
//! Multi-turn Astra scenarios snapshot the model-visible request history of shipped features.
use super::image_rollout::RecordingFileAttachmentStore;
use super::permissions_messages::PermissionPathScenario;
use codex_attachment_store::AttachmentStoreErrorKind;
use codex_protocol_test_support::ContentItemTestExt;
use std::collections::HashMap;
use std::collections::HashSet;
@@ -44,6 +46,8 @@ use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::AgentMessageDelivery;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::ImageReference;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::CodeModeToolMessages;
@@ -526,6 +530,161 @@ async fn realtime_attachment_replacement_preserves_handoff_context() -> Result<(
Ok(())
}
/// Expiring files change request copies while canonical history retains the original references.
/// A shared barrier requires the three distinct file IDs to resolve concurrently on each request.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unavailable_images_are_rechecked_without_rewriting_history() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let image_store = Arc::new(RecordingFileAttachmentStore {
resolve_barrier: Some(Arc::new(tokio::sync::Barrier::new(/*n*/ 3))),
..Default::default()
});
let test = test_codex()
.with_model("gpt-6-astra")
.with_image_store(image_store.clone())
.with_config(configure_scenario_catalog)
.build_with_auto_env(&server)
.await?;
let mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_assistant_message("first", "The images are available."),
ev_completed("first"),
]),
sse(vec![
ev_assistant_message("second", "Some images are no longer available."),
ev_completed("second"),
]),
],
)
.await;
let file_ids = [
"file_expiring",
"file_available",
"file_expiring",
"file_backend_error",
];
let mut input = vec![text("Compare these images.")];
input.extend(file_ids.map(|file_id| UserInput::Image {
image: ImageReference::File {
file_id: file_id.to_string(),
},
detail: Some(ImageDetail::High),
}));
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(input))
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(
*image_store.resolves.lock().expect("resolves lock"),
HashMap::from([
("file_expiring".to_string(), 1),
("file_available".to_string(), 1),
("file_backend_error".to_string(), 1),
])
);
image_store
.resolve_errors
.lock()
.expect("resolve errors lock")
.extend([
(
"file_expiring".to_string(),
AttachmentStoreErrorKind::NotFound,
),
(
"file_backend_error".to_string(),
AttachmentStoreErrorKind::Backend,
),
]);
test.submit_text_turn("Compare them again.").await?;
assert_eq!(
*image_store.resolves.lock().expect("resolves lock"),
HashMap::from([
("file_expiring".to_string(), 2),
("file_available".to_string(), 2),
("file_backend_error".to_string(), 2),
])
);
let requests = mock.requests();
assert_eq!(requests.len(), 2);
assert!(
requests[0]
.body_json()
.to_string()
.contains("file_expiring")
);
let unavailable_parts = requests[1]
.input()
.into_iter()
.filter_map(|item| {
item["internal_chat_message_metadata_passthrough"]["content_item_kinds"]
.as_array()
.cloned()
})
.flatten()
.filter(|kind| kind == "images.unavailable")
.count();
assert_eq!(unavailable_parts, 2);
assert!(requests[1].body_contains_text("[image unavailable]"));
assert!(
!requests[1]
.body_json()
.to_string()
.contains("file_expiring")
);
assert!(
requests[1]
.body_json()
.to_string()
.contains("file_backend_error")
);
assert!(
requests[1]
.body_json()
.to_string()
.contains("file_available")
);
let recorded_images = test
.codex
.load_history(/*include_archived*/ false)
.await?
.items
.into_iter()
.filter_map(|item| match item {
RolloutItem::ResponseItem(envelope) => match envelope.item {
ResponseItem::Message { content, .. } => Some(content),
_ => None,
},
_ => None,
})
.flatten()
.filter(|content| matches!(content, ContentItem::InputImage { .. }))
.collect::<Vec<_>>();
assert_eq!(
recorded_images,
file_ids.map(|file_id| ContentItem::InputImage {
image: ImageReference::File {
file_id: file_id.to_string()
},
detail: Some(ImageDetail::High),
})
);
insta::assert_snapshot!(context_snapshot::format_request_history_snapshot(
"Available file references pass through. A later request replaces missing files, preserves references on backend errors, rechecks each distinct ID once, and keeps canonical history intact.",
&requests,
&ContextSnapshotOptions::default().rewrite_known_segments(),
));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn code_mode_mcp_schema_limits_preserve_explicit_overrides() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -3023,6 +3182,109 @@ async fn async_compaction_test(
Ok((config_server, test))
}
/// Pending image checks for optional compaction do not delay foreground turn completion.
/// Its request still replaces files that expire after the sampling request.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn async_compaction_image_checks_do_not_block_turn_completion() -> Result<()> {
skip_if_no_network!(Ok(()));
let barrier = Arc::new(tokio::sync::Barrier::new(/*n*/ 2));
let image_store = Arc::new(RecordingFileAttachmentStore {
resolve_barrier: Some(Arc::clone(&barrier)),
..Default::default()
});
let (release_sampling, sampling_gate) = oneshot::channel();
let (streaming, _completions) = start_streaming_sse_server(vec![
vec![compaction_chunk(
Some(sampling_gate),
vec![
ev_assistant_message("answer", "I examined the image."),
ev_completed_with_tokens("answer", /*total_tokens*/ 70_000),
],
)],
vec![compaction_chunk(
/*gate*/ None,
vec![
compaction_output("IMAGE_CHECKPOINT"),
ev_completed_with_tokens("compact", /*total_tokens*/ 3_000),
],
)],
])
.await;
let config_server = start_mock_server().await;
let base_url = format!("{}/v1", streaming.uri());
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_image_store(image_store.clone())
.with_config(move |config| {
configure_scenario_catalog(config);
config.model_provider.base_url = Some(base_url);
config.model_context_window = Some(100_000);
config.features.enable(Feature::AsyncCompaction).unwrap();
config
.features
.disable(Feature::EnableRequestCompression)
.unwrap();
})
.with_model("gpt-6-astra")
.build_with_auto_env(&config_server)
.await?;
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![
text("Describe this image."),
UserInput::Image {
image: ImageReference::File {
file_id: "file_expiring".to_string(),
},
detail: Some(ImageDetail::High),
},
]))
.await?;
// Release the sampling lookup, then expire the file before its response finishes.
barrier.wait().await;
image_store
.resolve_errors
.lock()
.expect("resolve errors lock")
.insert(
"file_expiring".to_string(),
AttachmentStoreErrorKind::NotFound,
);
release_sampling.send(()).expect("finish sampling");
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(streaming.requests().await.len(), 1);
// The turn completed without releasing the optional compaction lookup.
barrier.wait().await;
tokio::time::timeout(
Duration::from_secs(/*secs*/ 20),
test.codex.wait_for_background_compaction_for_test(),
)
.await?;
let requests = streaming
.requests()
.await
.iter()
.map(|body| serde_json::from_slice::<serde_json::Value>(body))
.collect::<serde_json::Result<Vec<_>>>()?;
assert_eq!(requests.len(), 2);
assert_eq!(
*image_store.resolves.lock().expect("resolves lock"),
HashMap::from([("file_expiring".to_string(), 2)])
);
assert!(requests[0]["input"].to_string().contains("file_expiring"));
assert!(
requests[1]["input"]
.to_string()
.contains("[image unavailable]")
);
assert!(!requests[1]["input"].to_string().contains("file_expiring"));
test.codex.shutdown_and_wait().await?;
Ok(())
}
#[derive(Clone, Copy)]
enum LiveSuffix {
User,
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__unavailable_images_are_rechecked_without_rewriting_history.snap
+82
View File
@@ -0,0 +1,82 @@
---
source: core/tests/suite/scenarios.rs
assertion_line: 723
expression: "context_snapshot::format_request_history_snapshot(\"Available file references pass through. A later request replaces missing files, preserves references on backend errors, rechecks each distinct ID once, and keeps canonical history intact.\",\n&requests, &ContextSnapshotOptions::default().rewrite_known_segments(),)"
---
Scenario: Available file references pass through. A later request replaces missing files, preserves references on backend errors, rechecks each distinct ID once, and keeps canonical history intact.
## Window 1
-- request 1 (turn) --
00:additional_tools/developer (3; hash=bf2e30605ff5dc17):
- namespace/functions; hash=fada4a476a3754dd
- custom/exec
- function/wait
- function/request_user_input
- function/request_user_input_async
- namespace/clock: Tools for reading and waiting on time.; hash=f402e8c5e9b5e317
- function/sleep
- namespace/collaboration: Tools for spawning and managing sub-agents.; hash=2c8d6295978dbe63
- function/followup_task
- function/interrupt_agent
- function/list_agents
- function/send_message
- function/spawn_agent
- function/wait_agent
01:message/developer:
<MODEL_INSTRUCTIONS>
02:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <COLLABORATION_MODE>
03:message/developer:
<MULTI_AGENT_ROLE>
04:message/developer:
<MULTI_AGENT_MODE>
05:message/user:
<ENVIRONMENT_CONTEXT>
06:message/user[5]:
[01] Compare these images.
[02] <input_image:file_id>
[03] <input_image:file_id>
[04] <input_image:file_id>
[05] <input_image:file_id>
07:configuration_update:reasoning={"effort":"low"}
## Window 2 (after request 1: input diverged at item 06)
-- request 2 (turn) --
00:additional_tools/developer (3; hash=bf2e30605ff5dc17):
- namespace/functions; hash=fada4a476a3754dd
- custom/exec
- function/wait
- function/request_user_input
- function/request_user_input_async
- namespace/clock: Tools for reading and waiting on time.; hash=f402e8c5e9b5e317
- function/sleep
- namespace/collaboration: Tools for spawning and managing sub-agents.; hash=2c8d6295978dbe63
- function/followup_task
- function/interrupt_agent
- function/list_agents
- function/send_message
- function/spawn_agent
- function/wait_agent
01:message/developer:
<MODEL_INSTRUCTIONS>
02:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <COLLABORATION_MODE>
03:message/developer:
<MULTI_AGENT_ROLE>
04:message/developer:
<MULTI_AGENT_MODE>
05:message/user:
<ENVIRONMENT_CONTEXT>
06:message/user[5]:
[01] Compare these images.
[02] [image unavailable]
[03] <input_image:file_id>
[04] [image unavailable]
[05] <input_image:file_id>
07:configuration_update:reasoning={"effort":"low"}
08:message/assistant:
The images are available.
09:message/user:
Compare them again.