Add an injectable attachment store to ThreadManager (#42634)

## What changed

- Add the `codex-attachment-store` crate with storage-neutral attachment metadata, references, errors, and an asynchronous persistence interface.
- Provide an inline implementation that preserves attachment bytes as media-typed base64 data URLs.
- Inject the store into `ThreadManager`, expose it to consumers, and retain inline storage as the default for existing entry points.

## Testing

- Verify inline storage round-trips binary, text, PNG, and JPEG data.
- Verify attachment debug output redacts URLs while retaining file IDs.

GitOrigin-RevId: 7688dcd3c89d7540ed2398f3e7c63881fbfcda97
This commit is contained in:
Krish Chainani
2026-09-03 23:20:54 +00:00
committed by copyberry
parent d979df154c
commit 03467026f2
20 changed files with 249 additions and 0 deletions
+12
View File
@@ -2324,6 +2324,17 @@ dependencies = [
"tokio-util",
]
[[package]]
name = "codex-attachment-store"
version = "0.0.0"
dependencies = [
"base64 0.22.1",
"codex-utils-image",
"pretty_assertions",
"serde",
"tokio",
]
[[package]]
name = "codex-aws-auth"
version = "0.0.0"
@@ -2810,6 +2821,7 @@ dependencies = [
"codex-app-server-protocol",
"codex-apply-patch",
"codex-async-utils",
"codex-attachment-store",
"codex-client",
"codex-code-mode",
"codex-config",
+2
View File
@@ -9,6 +9,7 @@ members = [
"bwrap",
"build-info",
"ansi-escape",
"attachment-store",
"async-utils",
"app-server",
"app-server-transport",
@@ -166,6 +167,7 @@ codex-agent-identity = { path = "agent-identity" }
codex-agent-roles = { path = "agent-roles" }
codex-ansi-escape = { path = "ansi-escape" }
codex-api = { path = "codex-api" }
codex-attachment-store = { path = "attachment-store" }
codex-aws-auth = { path = "aws-auth" }
codex-app-server = { path = "app-server" }
codex-app-server-transport = { path = "app-server-transport" }
+1
View File
@@ -346,6 +346,7 @@ enabled = false
good_config.codex_home.clone(),
)),
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
Arc::clone(&thread_store),
codex_core::local_agent_graph_store_from_state_db(Some(&state_db)),
"11111111-1111-4111-8111-111111111111".to_string(),
@@ -348,6 +348,7 @@ impl MessageProcessor {
config.codex_home.clone(),
)),
Some(analytics_events_client.clone()),
codex_core::passthrough_image_store(),
Arc::clone(&thread_store),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "attachment-store",
crate_name = "codex_attachment_store",
)
+22
View File
@@ -0,0 +1,22 @@
[package]
edition.workspace = true
license.workspace = true
name = "codex-attachment-store"
version.workspace = true
[lib]
doctest = false
name = "codex_attachment_store"
path = "src/lib.rs"
[lints]
workspace = true
[dependencies]
codex-utils-image = { workspace = true }
serde = { workspace = true, features = ["derive"] }
[dev-dependencies]
base64 = { workspace = true }
pretty_assertions = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt"] }
+104
View File
@@ -0,0 +1,104 @@
//! Storage-neutral attachment persistence interfaces.
use std::error::Error;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use codex_utils_image::data_url_from_bytes;
use serde::Deserialize;
use serde::Serialize;
/// Result returned by [`AttachmentStore`] operations.
pub type AttachmentStoreResult<T> = Result<T, AttachmentStoreError>;
/// Future returned by [`AttachmentStore::persist`].
pub type AttachmentStoreFuture<'a> =
Pin<Box<dyn Future<Output = AttachmentStoreResult<AttachmentRef>> + Send + 'a>>;
/// Stores attachment bytes and returns durable references to them.
///
/// Implementations perform any required I/O before resolving the returned future. A successful
/// reference must remain valid when callers persist it or pass it across subsystem boundaries.
pub trait AttachmentStore: Send + Sync {
/// Persists `data` and returns a durable reference to it.
fn persist<'a>(
&'a self,
data: &'a [u8],
metadata: &'a AttachmentMetadata,
) -> AttachmentStoreFuture<'a>;
}
/// Metadata used to identify and interpret an attachment.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct AttachmentMetadata {
/// Name associated with the attachment.
pub file_name: String,
/// IANA media type used to interpret the attachment bytes.
pub media_type: String,
}
/// Durable reference to bytes stored by an [`AttachmentStore`].
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
pub struct AttachmentRef {
/// Identifier assigned to an externally stored file.
pub file_id: Option<String>,
/// URL referring to the attachment bytes, such as a data URL or
/// `sediment://<file_id>`.
pub url: String,
}
impl fmt::Debug for AttachmentRef {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("AttachmentRef")
.field("file_id", &self.file_id)
.field("url", &"<redacted>")
.finish()
}
}
/// Leaves attachments inline instead of storing them.
pub struct InlineAttachmentStore;
impl AttachmentStore for InlineAttachmentStore {
fn persist<'a>(
&'a self,
data: &'a [u8],
metadata: &'a AttachmentMetadata,
) -> AttachmentStoreFuture<'a> {
Box::pin(async move {
Ok(AttachmentRef {
file_id: None,
url: data_url_from_bytes(&metadata.media_type, data),
})
})
}
}
/// Error returned by an [`AttachmentStore`] implementation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct AttachmentStoreError {
message: String,
}
impl AttachmentStoreError {
/// Creates an error without exposing backend-specific error types.
pub fn new(message: impl Into<String>) -> Self {
Self {
message: message.into(),
}
}
}
impl fmt::Display for AttachmentStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.message.fmt(formatter)
}
}
impl Error for AttachmentStoreError {}
#[cfg(test)]
#[path = "lib_tests.rs"]
mod tests;
@@ -0,0 +1,60 @@
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use pretty_assertions::assert_eq;
use super::*;
/// Debug output retains the Sediment file ID while redacting credential-bearing URLs.
#[test]
fn attachment_ref_debug_redacts_url() {
let attachment = AttachmentRef {
file_id: Some("sediment-file-id".to_string()),
url: "https://attachments.test/file?signed=secret".to_string(),
};
assert_eq!(
format!("{attachment:?}"),
r#"AttachmentRef { file_id: Some("sediment-file-id"), url: "<redacted>" }"#
);
}
/// The inline store preserves binary, text, PNG, and JPEG attachment bytes.
#[tokio::test]
async fn inline_store_preserves_attachment_bytes() {
let cases: [(&str, &str, &[u8]); 4] = [
(
"attachment.bin",
"application/octet-stream",
b"\x00\x01\x7f\x80\xfe\xff",
),
("note.txt", "text/plain", b"hello\n"),
("image.png", "image/png", b"\x89PNG\r\n\x1a\n"),
(
"image.jpg",
"image/jpeg",
b"\xff\xd8\xff\xe0JFIF\x00\xff\xd9",
),
];
for (file_name, media_type, data) in cases {
let metadata = AttachmentMetadata {
file_name: file_name.to_string(),
media_type: media_type.to_string(),
};
let attachment = InlineAttachmentStore
.persist(data, &metadata)
.await
.expect("inline attachment");
let (data_url_metadata, encoded) =
attachment.url.split_once(',').expect("data URL payload");
let decoded = BASE64_STANDARD
.decode(encoded)
.expect("valid base64 payload");
let expected_data_url_metadata = format!("data:{media_type};base64");
assert_eq!(
(attachment.file_id, data_url_metadata, decoded),
(None, expected_data_url_metadata.as_str(), data.to_vec())
);
}
}
+1
View File
@@ -67,6 +67,7 @@ pub use codex_core::config::ThreadStoreConfig;
pub use codex_core::config::find_codex_home;
pub use codex_core::init_state_db;
pub use codex_core::local_agent_graph_store_from_state_db;
pub use codex_core::passthrough_image_store;
pub use codex_core::resolve_installation_id;
pub use codex_core::thread_store_from_config;
pub use codex_exec_server::EnvironmentManager;
+1
View File
@@ -25,6 +25,7 @@ codex-api = { workspace = true }
codex-app-server-protocol = { workspace = true }
codex-apply-patch = { workspace = true }
codex-async-utils = { workspace = true }
codex-attachment-store = { workspace = true }
codex-client = { workspace = true }
codex-code-mode = { workspace = true }
codex-connectors = { workspace = true }
+1
View File
@@ -3639,6 +3639,7 @@ async fn resume_thread_subagent_restores_stored_metadata() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
crate::thread_manager::passthrough_image_store(),
thread_store.clone(),
/*agent_graph_store*/ None,
uuid::Uuid::new_v4().to_string(),
+1
View File
@@ -125,6 +125,7 @@ pub use thread_manager::ThreadManager;
pub use thread_manager::ThreadShutdownReport;
pub use thread_manager::build_models_manager;
pub use thread_manager::local_agent_graph_store_from_state_db;
pub use thread_manager::passthrough_image_store;
pub use thread_manager::thread_store_from_config;
pub use tools::handlers::WaitForEnvironmentToolConfig;
pub use web_search::web_search_action_detail;
+1
View File
@@ -62,6 +62,7 @@ pub async fn build_prompt_input(
extensions,
user_instructions_provider,
/*analytics_events_client*/ None,
crate::thread_manager::passthrough_image_store(),
thread_store,
crate::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
+15
View File
@@ -23,6 +23,8 @@ use codex_agent_graph_store::LocalAgentGraphStore;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::TurnStatus;
use codex_attachment_store::AttachmentStore;
use codex_attachment_store::InlineAttachmentStore;
use codex_code_mode::CodeModeSessionProvider;
use codex_code_mode::DisabledCodeModeSessionProvider;
use codex_code_mode::ProcessOwnedCodeModeSessionProvider;
@@ -355,6 +357,7 @@ pub(crate) struct ThreadManagerState {
code_mode_session_provider: Arc<dyn CodeModeSessionProvider>,
extensions: Arc<ExtensionRegistry<Config>>,
user_instructions_provider: Arc<dyn UserInstructionsProvider>,
image_store: Arc<dyn AttachmentStore>,
thread_store: Arc<dyn ThreadStore>,
agent_graph_store: Option<Arc<dyn AgentGraphStore>>,
attestation_provider: Option<Arc<dyn AttestationProvider>>,
@@ -414,6 +417,11 @@ pub fn thread_store_from_config(
}
}
/// Constructs the default image store that preserves inline images.
pub fn passthrough_image_store() -> Arc<dyn AttachmentStore> {
Arc::new(InlineAttachmentStore)
}
/// Construct the default SQLite-backed agent graph store when local state is available.
pub fn local_agent_graph_store_from_state_db(
state_db: Option<&StateDbHandle>,
@@ -435,6 +443,7 @@ impl ThreadManager {
extensions: Arc<ExtensionRegistry<Config>>,
user_instructions_provider: Arc<dyn UserInstructionsProvider>,
analytics_events_client: Option<AnalyticsEventsClient>,
image_store: Arc<dyn AttachmentStore>,
thread_store: Arc<dyn ThreadStore>,
agent_graph_store: Option<Arc<dyn AgentGraphStore>>,
installation_id: String,
@@ -483,6 +492,7 @@ impl ThreadManager {
code_mode_session_provider,
extensions,
user_instructions_provider,
image_store,
thread_store,
agent_graph_store,
attestation_provider,
@@ -633,6 +643,7 @@ impl ThreadManager {
user_instructions_provider: Arc::new(
crate::test_support::EmptyUserInstructionsProvider,
),
image_store: passthrough_image_store(),
thread_store,
agent_graph_store,
attestation_provider: None,
@@ -672,6 +683,10 @@ impl ThreadManager {
self.state.environment_manager.clone()
}
pub fn image_store(&self) -> Arc<dyn AttachmentStore> {
Arc::clone(&self.state.image_store)
}
/// Starts the local rollout migration path after a runtime feature enablement.
///
/// Startup config handles the initial launch in [`thread_store_from_config`]. This covers
+15
View File
@@ -805,6 +805,7 @@ async fn mcp_invalidation_refreshes_threads_that_are_still_starting() {
Arc::new(extensions.build()),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -1303,6 +1304,7 @@ async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors()
Arc::new(extensions.build()),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -1510,6 +1512,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -1654,6 +1657,7 @@ async fn explicit_installation_id_skips_codex_home_file() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store,
local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id.clone(),
@@ -1697,6 +1701,7 @@ async fn resume_active_thread_from_rollout_returns_running_thread() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -1759,6 +1764,7 @@ async fn resume_stopped_thread_from_rollout_spawns_new_thread() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -1828,6 +1834,7 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store,
local_agent_graph_store_from_state_db(state_db.as_ref()),
TEST_INSTALLATION_ID.to_string(),
@@ -1913,6 +1920,7 @@ async fn subtree_listing_uses_injected_graph_store_without_state_db() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
Some(agent_graph_store),
TEST_INSTALLATION_ID.to_string(),
@@ -1960,6 +1968,7 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store.clone(),
local_agent_graph_store_from_state_db(state_db.as_ref()),
TEST_INSTALLATION_ID.to_string(),
@@ -2070,6 +2079,7 @@ async fn metadata_update_without_result_reads_only_when_the_caller_needs_the_thr
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store.clone(),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -2196,6 +2206,7 @@ async fn new_uses_active_provider_for_model_refresh() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -2243,6 +2254,7 @@ async fn injected_models_manager_controls_refresh_policy() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
TEST_INSTALLATION_ID.to_string(),
@@ -2503,6 +2515,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, state_db.clone()),
local_agent_graph_store_from_state_db(state_db.as_ref()),
TEST_INSTALLATION_ID.to_string(),
@@ -2616,6 +2629,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, state_db.clone()),
local_agent_graph_store_from_state_db(state_db.as_ref()),
TEST_INSTALLATION_ID.to_string(),
@@ -2719,6 +2733,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
passthrough_image_store(),
thread_store_from_config(&config, state_db.clone()),
local_agent_graph_store_from_state_db(state_db.as_ref()),
TEST_INSTALLATION_ID.to_string(),
@@ -4153,6 +4153,7 @@ async fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtr
empty_extension_registry(),
Arc::new(crate::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
crate::thread_manager::passthrough_image_store(),
thread_store_from_config(&config, state_db.clone()),
local_agent_graph_store_from_state_db(state_db.as_ref()),
"11111111-1111-4111-8111-111111111111".to_string(),
+1
View File
@@ -702,6 +702,7 @@ impl TestCodexBuilder {
Arc::clone(&self.extensions),
user_instructions_provider,
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
Arc::clone(&thread_store),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
+1
View File
@@ -1805,6 +1805,7 @@ async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() {
empty_extension_registry(),
Arc::new(codex_core::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
installation_id,
@@ -106,6 +106,7 @@ impl MessageProcessor {
Arc::new(extensions.build()),
user_instructions_provider,
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
codex_core::thread_store_from_config(config.as_ref(), state_db.clone()),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
@@ -66,6 +66,7 @@ use codex_core_api::init_state_db;
use codex_core_api::install_image_generation_extension;
use codex_core_api::item_event_to_server_notification;
use codex_core_api::local_agent_graph_store_from_state_db;
use codex_core_api::passthrough_image_store;
use codex_core_api::resolve_installation_id;
use codex_core_api::set_default_originator;
use codex_core_api::thread_store_from_config;
@@ -149,6 +150,7 @@ async fn run_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
Arc::new(extensions.build()),
user_instructions_provider,
/*analytics_events_client*/ None,
passthrough_image_store(),
Arc::clone(&thread_store),
local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,