From 19286b88190b1b81c2506af97dfca6ab261ff72b Mon Sep 17 00:00:00 2001 From: joeytrasatti-openai Date: Tue, 15 Sep 2026 01:46:46 +0000 Subject: [PATCH] Copy current thread attachments into non-ephemeral forks (#45579) ## What changed Copy the source thread's current attachments when creating a non-ephemeral fork, including forks at an earlier turn. Copies receive new attachment IDs and creation timestamps while preserving resource identities and payloads. Attachment membership can then change independently on either thread; referenced resources are not copied. Await the atomic copy before publishing the fork. If copying fails, log the error and allow the conversation fork to succeed without attachments. Resuming a fork does not copy attachments again. Document that clients should use `forkedFromId` on `thread/started` and fetch the fork's attachments with `thread/attachment/list`; copying does not emit per-attachment updates. ## Testing Add coverage for atomic rollback, independent attachment membership, inheritance across history cutoffs, successful conversation forks after copy failures, and resuming without restoring removed attachments. GitOrigin-RevId: 939d2c6a3073ccfcd922380c6beda45d3deac606 --- codex-rs/Cargo.lock | 1 + codex-rs/app-server/Cargo.toml | 1 + codex-rs/app-server/README.md | 2 + .../app-server/tests/suite/v2/thread_fork.rs | 178 ++++++++++++++++-- codex-rs/core/src/thread_manager.rs | 17 ++ .../state/src/runtime/thread_attachments.rs | 40 ++++ .../src/runtime/thread_attachments_tests.rs | 66 +++++++ codex-rs/thread-store/src/local/mod.rs | 15 ++ .../src/local/thread_attachments.rs | 18 ++ codex-rs/thread-store/src/store.rs | 16 ++ 10 files changed, 339 insertions(+), 15 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9c9c1a6ec1..26a070df9a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2254,6 +2254,7 @@ dependencies = [ "serial_test", "sha2 0.10.9", "shlex", + "sqlx", "tar", "tempfile", "test-case", diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 0a30f470ef..5c43378311 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -147,6 +147,7 @@ rmcp = { workspace = true, default-features = false, features = [ ] } serial_test = { workspace = true } shlex = { workspace = true } +sqlx = { workspace = true } tar = { workspace = true } test-case = "3.3.1" tokio-tungstenite = { workspace = true } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 358ae55943..3af8716a16 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -201,6 +201,8 @@ Attachments record the resources currently associated with a thread, independent `thread/attachment/list` accepts one `threadId` and returns at most 100 attachments per page, ordered by creation time and attachment id. Continue with `nextCursor` and the same `threadId` until the cursor is `null`. Each thread can retain up to 100 attachments. Removing an attachment frees a slot for a new attachment. +A non-ephemeral fork copies the source thread's current attachments, even when forking at an earlier turn. The copies have new attachment IDs and creation timestamps, but retain the same resource identities and payloads. Clients use `forkedFromId` on `thread/started` to detect forks and call `thread/attachment/list` with the new thread ID to load their attachments. Fork copying does not emit per-attachment updates; explicit add/remove operations still do. Copying is awaited before publishing the fork, but is best effort: a copy failure is logged and the conversation fork succeeds without attachments. Membership can then change independently on either thread; the referenced resources themselves are not copied. Resuming a fork does not repeat the copy. + Attachment creation and deletion requests using the same thread ID are serialized across connections. The requesting client receives its response before the compact update is broadcast, and duplicate creates or absent deletes do not emit updates. Deleting the owning thread removes its attachments under the same lifecycle exclusion; queued attachment mutations then report that the thread was not found. # Thread plugin settings diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 0253646ae5..3690b3eb1f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -23,6 +23,12 @@ use codex_app_server_protocol::SandboxMode; use codex_app_server_protocol::SandboxPolicy; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SessionSource; +use codex_app_server_protocol::ThreadAttachmentAddParams; +use codex_app_server_protocol::ThreadAttachmentAddResponse; +use codex_app_server_protocol::ThreadAttachmentListParams; +use codex_app_server_protocol::ThreadAttachmentListResponse; +use codex_app_server_protocol::ThreadAttachmentRemoveParams; +use codex_app_server_protocol::ThreadAttachmentRemoveResponse; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; @@ -623,6 +629,75 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( } let original_contents = std::fs::read_to_string(source_path.as_path())?; + // Membership added after the requested history cutoff must still be inherited. + let attachment: ThreadAttachmentAddResponse = mcp + .request(|request_id| ClientRequest::ThreadAttachmentAdd { + request_id, + params: ThreadAttachmentAddParams { + thread_id: source_thread_id.clone(), + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#123".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/123"}), + }, + }) + .await?; + let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()); + let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + sqlx::query( + "CREATE TRIGGER fail_fork_attachment_copy BEFORE INSERT ON thread_attachments \ + BEGIN SELECT RAISE(FAIL, 'attachment copy failure'); END", + ) + .execute(&pool) + .await?; + let fork_without_attachments: ThreadForkResponse = mcp + .request(|request_id| ClientRequest::ThreadFork { + request_id, + params: ThreadForkParams { + thread_id: source_thread_id.clone(), + last_turn_id: Some(turn_ids[1].clone()), + ..Default::default() + }, + }) + .await?; + assert_eq!( + fork_without_attachments + .thread + .turns + .iter() + .map(|turn| turn.id.clone()) + .collect::>(), + turn_ids[..2] + ); + assert!( + fork_without_attachments + .thread + .path + .as_ref() + .expect("durable fork path") + .exists() + ); + let without_attachments: ThreadAttachmentListResponse = mcp + .request(|request_id| ClientRequest::ThreadAttachmentList { + request_id, + params: ThreadAttachmentListParams { + thread_id: fork_without_attachments.thread.id.clone(), + cursor: None, + limit: None, + }, + }) + .await?; + assert!(without_attachments.data.is_empty()); + assert_eq!( + sqlx::query_scalar::<_, String>("SELECT id FROM threads WHERE id = ?") + .bind(&fork_without_attachments.thread.id) + .fetch_one(&pool) + .await?, + fork_without_attachments.thread.id + ); + sqlx::query("DROP TRIGGER fail_fork_attachment_copy") + .execute(&pool) + .await?; + pool.close().await; let fork_id = mcp .send_thread_fork_request(ThreadForkParams { thread_id: source_thread_id.clone(), @@ -635,6 +710,66 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??; + let started = loop { + let notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/started"), + ) + .await??; + let started: ThreadStartedNotification = + serde_json::from_value(notification.params.expect("params must be present"))?; + if started.thread.id == forked_thread.id { + break started; + } + }; + assert!(started.thread.turns.is_empty()); + assert_eq!( + started.thread.forked_from_id, + Some(source_thread_id.clone()) + ); + + let copied: ThreadAttachmentListResponse = mcp + .request(|request_id| ClientRequest::ThreadAttachmentList { + request_id, + params: ThreadAttachmentListParams { + thread_id: started.thread.id, + cursor: None, + limit: None, + }, + }) + .await?; + assert_eq!(copied.data.len(), 1); + assert_ne!(copied.data[0].id, attachment.attachment.id); + assert_eq!( + copied.data[0], + codex_app_server_protocol::ThreadAttachment { + id: copied.data[0].id.clone(), + created_at: copied.data[0].created_at, + ..attachment.attachment.clone() + } + ); + let _: ThreadAttachmentRemoveResponse = mcp + .request(|request_id| ClientRequest::ThreadAttachmentRemove { + request_id, + params: ThreadAttachmentRemoveParams { + thread_id: forked_thread.id.clone(), + attachment_type: attachment.attachment.attachment_type.clone(), + identity_key: attachment.attachment.identity_key.clone(), + }, + }) + .await?; + let parent_attachments: ThreadAttachmentListResponse = mcp + .request(|request_id| ClientRequest::ThreadAttachmentList { + request_id, + params: ThreadAttachmentListParams { + thread_id: source_thread_id.clone(), + cursor: None, + limit: None, + }, + }) + .await?; + assert_eq!(parent_attachments.data, vec![attachment.attachment]); + assert_eq!( forked_thread .turns @@ -675,20 +810,6 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( } assert!(!forked_contents.contains(turn_ids[2].as_str())); - let started = loop { - let notification = timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("thread/started"), - ) - .await??; - let started: ThreadStartedNotification = - serde_json::from_value(notification.params.expect("params must be present"))?; - if started.thread.id == forked_thread.id { - break started; - } - }; - assert!(started.thread.turns.is_empty()); - if history_mode == ThreadHistoryMode::Paginated { let before_fork_id = mcp .send_thread_fork_request(ThreadForkParams { @@ -729,7 +850,7 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( .request(|request_id| ClientRequest::ThreadFork { request_id, params: ThreadForkParams { - thread_id: forked_thread.id, + thread_id: forked_thread.id.clone(), before_turn_id: Some(completed.turn.id), ephemeral: true, exclude_turns: true, @@ -740,6 +861,33 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( assert_eq!(ephemeral_fork.preview, "first"); } + // Resuming a fork must not copy its parent's attachments again after an explicit removal. + mcp.shutdown_gracefully().await?; + let mut resumed = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let _: ThreadResumeResponse = resumed + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: forked_thread.id.clone(), + ..Default::default() + }, + }) + .await?; + let after_resume: ThreadAttachmentListResponse = resumed + .request(|request_id| ClientRequest::ThreadAttachmentList { + request_id, + params: ThreadAttachmentListParams { + thread_id: forked_thread.id, + cursor: None, + limit: None, + }, + }) + .await?; + assert!(after_resume.data.is_empty()); + Ok(()) } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 5b3cd9b299..ebfb570766 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -2145,6 +2145,8 @@ impl ThreadManagerState { } else { codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile }; + let attachment_source = + forked_from_thread_id.filter(|_| matches!(&initial_history, InitialHistory::Forked(_))); let (session, io) = Session::spawn(SessionSpawnArgs { startup, config, @@ -2207,6 +2209,21 @@ impl ThreadManagerState { windows_sandbox_proxy_settings_mode, }) .await?; + if let Some(source_thread_id) = attachment_source + && session.live_thread().is_some() + && self.thread_store.supports_thread_attachments() + { + // Fork initialization already handles persistence. Copy current membership before + // registration, but do not fail the conversation fork for attachment metadata errors. + let thread_id = session.thread_id(); + if let Err(error) = self + .thread_store + .copy_thread_attachments(source_thread_id, thread_id) + .await + { + warn!(%thread_id, %source_thread_id, %error, "failed to copy fork attachments"); + } + } // Enable Full Access form input only after session startup so a required MCP server cannot // block startup while waiting for form input. if session diff --git a/codex-rs/state/src/runtime/thread_attachments.rs b/codex-rs/state/src/runtime/thread_attachments.rs index 7e6cfddcb0..d581250167 100644 --- a/codex-rs/state/src/runtime/thread_attachments.rs +++ b/codex-rs/state/src/runtime/thread_attachments.rs @@ -21,6 +21,46 @@ use sqlx::sqlite::SqliteRow; use uuid::Uuid; impl StateRuntime { + /// Atomically copies current membership into a new, empty fork, independent of history cutoffs. + pub async fn copy_thread_attachments( + &self, + source_thread_id: ThreadId, + destination_thread_id: ThreadId, + ) -> anyhow::Result<()> { + let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; + let destination = destination_thread_id.to_string(); + let exists = sqlx::query_scalar::<_, i64>("SELECT 1 FROM threads WHERE id = ?") + .bind(&destination) + .fetch_optional(&mut *transaction) + .await? + .is_some(); + if !exists { + anyhow::bail!("thread not found: {destination_thread_id}"); + } + let rows = sqlx::query( + "SELECT attachment_type, identity_key, payload FROM thread_attachments WHERE thread_id = ? ORDER BY created_at, id", + ) + .bind(source_thread_id.to_string()) + .fetch_all(&mut *transaction) + .await?; + let created_at = Utc::now().timestamp(); + for row in rows { + sqlx::query( + "INSERT INTO thread_attachments (id, thread_id, attachment_type, identity_key, payload, created_at) VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(Uuid::now_v7().to_string()) + .bind(&destination) + .bind(row.try_get::("attachment_type")?) + .bind(row.try_get::("identity_key")?) + .bind(row.try_get::("payload")?) + .bind(created_at) + .execute(&mut *transaction) + .await?; + } + transaction.commit().await?; + Ok(()) + } + /// Attach an attachment once, returning an existing attachment for repeated requests. pub async fn add_thread_attachment( &self, diff --git a/codex-rs/state/src/runtime/thread_attachments_tests.rs b/codex-rs/state/src/runtime/thread_attachments_tests.rs index 7bbcb6be84..0cf8f05d0c 100644 --- a/codex-rs/state/src/runtime/thread_attachments_tests.rs +++ b/codex-rs/state/src/runtime/thread_attachments_tests.rs @@ -40,6 +40,72 @@ async fn runtime_with_threads(count: usize) -> Result<(Arc, PathBu Ok((runtime, codex_home, thread_ids)) } +#[tokio::test] +async fn copying_attachments_is_atomic_and_independent() -> Result<()> { + let (runtime, _codex_home, thread_ids) = runtime_with_threads(/*count*/ 2).await?; + let source = thread_ids[0]; + let destination = thread_ids[1]; + for key in ["first", "second"] { + runtime + .add_thread_attachment(source, "document", key, &json!({"path": key})) + .await?; + } + sqlx::query( + "CREATE TRIGGER fail_attachment_copy BEFORE INSERT ON thread_attachments WHEN NEW.identity_key = 'second' BEGIN SELECT RAISE(FAIL, 'copy failure'); END", + ) + .execute(runtime.pool.as_ref()) + .await?; + let error = runtime + .copy_thread_attachments(source, destination) + .await + .expect_err("a failed insert must roll back the entire copy"); + assert!(error.to_string().contains("copy failure")); + assert!( + runtime + .list_thread_attachments(destination, /*cursor*/ None, /*limit*/ 10) + .await? + .attachments + .is_empty() + ); + sqlx::query("DROP TRIGGER fail_attachment_copy") + .execute(runtime.pool.as_ref()) + .await?; + runtime.copy_thread_attachments(source, destination).await?; + let originals = runtime + .list_thread_attachments(source, /*cursor*/ None, /*limit*/ 10) + .await? + .attachments; + let copied = runtime + .list_thread_attachments(destination, /*cursor*/ None, /*limit*/ 10) + .await? + .attachments; + assert_eq!(copied.len(), originals.len()); + for (original, copy) in originals.iter().zip(&copied) { + assert_ne!(copy.id, original.id); + assert_eq!( + copy, + &crate::ThreadAttachment { + id: copy.id.clone(), + thread_id: destination, + created_at: copy.created_at, + ..original.clone() + } + ); + } + runtime + .remove_thread_attachment(source, "document", "first") + .await?; + runtime.delete_thread(source).await?; + assert_eq!( + runtime + .list_thread_attachments(destination, /*cursor*/ None, /*limit*/ 10) + .await? + .attachments, + copied + ); + Ok(()) +} + #[tokio::test] async fn attachment_attachments_are_idempotent_and_scoped_to_their_thread() -> Result<()> { let (runtime, _codex_home, thread_ids) = runtime_with_threads(/*count*/ 2).await?; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index de8f12a041..0e170c2439 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -604,6 +604,21 @@ impl ThreadStore for LocalThreadStore { self.state_db.is_some() } + fn copy_thread_attachments( + &self, + source_thread_id: ThreadId, + destination_thread_id: ThreadId, + ) -> ThreadStoreFuture<'_, ()> { + Box::pin(async move { + thread_attachments::copy_thread_attachments( + self, + source_thread_id, + destination_thread_id, + ) + .await + }) + } + fn add_thread_attachment( &self, params: AddThreadAttachmentParams, diff --git a/codex-rs/thread-store/src/local/thread_attachments.rs b/codex-rs/thread-store/src/local/thread_attachments.rs index 95cea009c4..aa20055207 100644 --- a/codex-rs/thread-store/src/local/thread_attachments.rs +++ b/codex-rs/thread-store/src/local/thread_attachments.rs @@ -12,6 +12,24 @@ use crate::ThreadStoreResult; use codex_protocol::ThreadId; use codex_rollout::StateDbHandle; +pub(super) async fn copy_thread_attachments( + store: &LocalThreadStore, + source_thread_id: ThreadId, + destination_thread_id: ThreadId, +) -> ThreadStoreResult<()> { + let state_db = state_db(store, "copy_thread_attachments")?; + let _lifecycle_reservation = store + .live_writer_locks + .reserve_lifecycle(destination_thread_id) + .await; + // Reference-backed forks already reserve their source. Do not recursively acquire that + // reservation: a queued deletion could deadlock it. SQLite serializes the source snapshot. + state_db + .copy_thread_attachments(source_thread_id, destination_thread_id) + .await + .map_err(|error| attachment_error("copy", Some(destination_thread_id), error)) +} + pub(super) async fn add_thread_attachment( store: &LocalThreadStore, params: AddThreadAttachmentParams, diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index d43b69447c..58c54db156 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -285,6 +285,22 @@ pub trait ThreadStore: Any + Send + Sync { false } + /// Copies current attachment membership into a newly persisted fork. + /// + /// Copies must be atomic and use new attachment IDs. The destination must be empty; + /// subsequent membership changes on either thread must remain independent. + fn copy_thread_attachments( + &self, + _source_thread_id: ThreadId, + _destination_thread_id: ThreadId, + ) -> ThreadStoreFuture<'_, ()> { + Box::pin(async { + Err(ThreadStoreError::Unsupported { + operation: "copy_thread_attachments", + }) + }) + } + /// Attaches an attachment, returning an existing attachment for repeated requests. fn add_thread_attachment( &self,