openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-13 21:30:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Resolve queued attachment paths against the thread working directory (#61359)

Browse Source 
## Why

Relative attachment paths in queued submissions should refer to files in the thread's working directory, even when the app server runs from a different directory.

## What changed

Resolve relative `LocalImage` and `LocalAudio` paths against the loaded thread's current working directory or the stored working directory for an unloaded thread when adding or updating queued submissions. Preserve absolute paths and reject relative paths that cannot be resolved to an absolute path.

## Testing

Add a regression test for queue add and update with different images sharing the same filename in the thread and server working directories. Verify that the relative path selects the thread's image and that explicit absolute paths retain their meaning.

GitOrigin-RevId: 3c1a48693c031f1c650650a9b3a7059fb962ba5c
This commit is contained in:
Eric Traut
2026-11-13 18:53:48 +00:00
committed by copyberry
parent dd8cee9822
commit 4413deac57
2 changed files with 104 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_queue_processor.rs
+45 -17
View File
@@ -1,4 +1,6 @@
use crate::config_manager::ConfigManager;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use crate::error_code::internal_error;
@@ -79,13 +81,18 @@ impl ThreadQueueRequestProcessor {
params: ThreadQueueAddParams,
) -> Result<ThreadQueueAddResponse, JSONRPCErrorError> {
validate_user_input_image_urls(&params.input)?;
let (thread_id, loaded_thread, source) = self.require_thread(&params.thread_id).await?;
let (thread_id, loaded_thread, source, cwd) =
self.require_thread(&params.thread_id).await?;
ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
let queued_item = self
.service()?
.enqueue_with_metadata(
thread_id,
submission_into_turn_input(params.input, Some(params.client_user_message_id)),
submission_into_turn_input(
params.input,
Some(params.client_user_message_id),
&cwd,
)?,
params.responsesapi_client_metadata,
)
.await
@@ -99,7 +106,7 @@ impl ThreadQueueRequestProcessor {
&self,
params: ThreadQueueListParams,
) -> Result<ThreadQueueListResponse, JSONRPCErrorError> {
let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
let (thread_id, ..) = self.require_thread(&params.thread_id).await?;
let offset = params
.cursor
.as_deref()
@@ -137,14 +144,19 @@ impl ThreadQueueRequestProcessor {
params: ThreadQueueUpdateParams,
) -> Result<ThreadQueueUpdateResponse, JSONRPCErrorError> {
validate_user_input_image_urls(&params.input)?;
let (thread_id, loaded_thread, source) = self.require_thread(&params.thread_id).await?;
let (thread_id, loaded_thread, source, cwd) =
self.require_thread(&params.thread_id).await?;
ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
let queued_item = self
.service()?
.update_with_metadata(
thread_id,
params.queued_submission_id.clone(),
submission_into_turn_input(params.input, /*client_user_message_id*/ None),
submission_into_turn_input(
params.input,
/*client_user_message_id*/ None,
&cwd,
)?,
params.responsesapi_client_metadata,
)
.await
@@ -164,7 +176,7 @@ impl ThreadQueueRequestProcessor {
&self,
params: ThreadQueueDeleteParams,
) -> Result<ThreadQueueDeleteResponse, JSONRPCErrorError> {
let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
let (thread_id, ..) = self.require_thread(&params.thread_id).await?;
let deleted = self
.service()?
.delete(thread_id, params.queued_submission_id)
@@ -177,7 +189,7 @@ impl ThreadQueueRequestProcessor {
&self,
params: ThreadQueueReorderParams,
) -> Result<ThreadQueueReorderResponse, JSONRPCErrorError> {
let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
let (thread_id, ..) = self.require_thread(&params.thread_id).await?;
self.service()?
.reorder(thread_id, params.queued_submission_ids)
.await
@@ -190,7 +202,7 @@ impl ThreadQueueRequestProcessor {
request_id: &ConnectionRequestId,
params: ThreadQueueStartParams,
) -> Result<ThreadQueueStartResponse, JSONRPCErrorError> {
let (_, loaded_thread, source) = self.require_thread(&params.thread_id).await?;
let (_, loaded_thread, source, _) = self.require_thread(&params.thread_id).await?;
ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
let thread = loaded_thread
.ok_or_else(|| invalid_request("resume the thread before starting a queued message"))?;
@@ -255,10 +267,11 @@ impl ThreadQueueRequestProcessor {
async fn require_thread(
&self,
raw_thread_id: &str,
) -> Result<(ThreadId, Option<Arc<CodexThread>>, SessionSource), JSONRPCErrorError> {
) -> Result<(ThreadId, Option<Arc<CodexThread>>, SessionSource, PathBuf), JSONRPCErrorError>
{
let thread_id = ThreadId::from_string(raw_thread_id)
.map_err(|error| invalid_request(format!("invalid thread id: {error}")))?;
let (loaded_thread, source) = if let Ok(thread) =
let (loaded_thread, source, cwd) = if let Ok(thread) =
self.thread_manager.get_thread(thread_id).await
{
let snapshot = thread.config_snapshot().await;
@@ -267,7 +280,8 @@ impl ThreadQueueRequestProcessor {
"ephemeral thread does not support queued submissions: {thread_id}"
)));
}
(Some(thread), snapshot.session_source)
let cwd = snapshot.cwd().to_path_buf();
(Some(thread), snapshot.session_source, cwd)
} else {
let stored = self
.thread_store
@@ -288,10 +302,10 @@ impl ThreadQueueRequestProcessor {
"session {thread_id} is archived. Run `codex unarchive {thread_id}` to unarchive it first."
)));
}
(None, stored.source)
(None, stored.source, stored.cwd)
};
Ok((thread_id, loaded_thread, source))
Ok((thread_id, loaded_thread, source, cwd))
}
}
@@ -315,13 +329,27 @@ fn ensure_direct_input_allowed(
}
fn submission_into_turn_input(
input: Vec<UserInput>,
mut input: Vec<UserInput>,
client_user_message_id: Option<String>,
) -> TurnInput {
TurnInput::UserInput {
cwd: &Path,
) -> Result<TurnInput, JSONRPCErrorError> {
for item in &mut input {
if let UserInput::LocalImage { path, .. } | UserInput::LocalAudio { path } = item
&& path.is_relative()
{
let resolved = cwd.join(&*path);
if !resolved.is_absolute() {
return Err(invalid_request(
"local queued attachment path cannot be resolved against the thread working directory",
));
}
*path = resolved;
}
}
Ok(TurnInput::UserInput {
content: input.into_iter().map(UserInput::into_core).collect(),
client_id: client_user_message_id,
}
})
}
pub(super) fn queue_error(error: QueueServiceError) -> JSONRPCErrorError {
codex-rs/app-server/tests/suite/v2/thread_queue.rs
+59
View File
@@ -9,6 +9,8 @@ use app_test_support::create_escalated_command_execution_sse_response;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
@@ -270,6 +272,63 @@ async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn queue_attachments_resolve_relative_paths_against_thread_cwd() -> Result<()> {
skip_if_remote!(
Ok(()),
"uses app-server-local attachments and a local thread cwd"
);
const RED_PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC";
const BLUE_PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGNgYPgPAAEDAQAIicLsAAAAAElFTkSuQmCC";
let (mut app, codex_home, _server) = queue_app(vec![blocked_turn_response()?]).await?;
let workspace = TempDir::new()?;
// The test app-server starts in CODEX_HOME.
let server_image = codex_home.path().join("sample.png");
let workspace_image = workspace.path().join("sample.png");
std::fs::write(&server_image, STANDARD.decode(RED_PNG)?)?;
std::fs::write(&workspace_image, STANDARD.decode(BLUE_PNG)?)?;
let start_id = app
.send_thread_start_request(ThreadStartParams {
cwd: Some(workspace.path().display().to_string()),
..Default::default()
})
.await?;
let started: ThreadStartResponse = timeout(READ_TIMEOUT, app.read_response(start_id)).await??;
let thread_id = started.thread.id;
let _ = start_blocked_turn(&mut app, &thread_id).await?;
let input: Vec<_> = ["sample.png".into(), workspace_image, server_image]
.into_iter()
.map(|path| UserInput::LocalImage { path, detail: None })
.collect();
let queued = queue_item(
&mut app,
ThreadQueueAddParams {
input: input.clone(),
..submission(&thread_id, "attachment")
},
)
.await?;
assert_eq!(queued.input[0], queued.input[1]);
assert_ne!(queued.input[0], queued.input[2]);
let updated: ThreadQueueUpdateResponse = app
.request(|request_id| ClientRequest::ThreadQueueUpdate {
request_id,
params: ThreadQueueUpdateParams {
thread_id,
queued_submission_id: queued.id,
input,
responsesapi_client_metadata: None,
},
})
.await?;
assert_eq!(updated.queued_submission.input, queued.input);
Ok(())
}
#[tokio::test]
async fn queue_list_returns_ordered_pages_and_lightweight_notifications() -> Result<()> {
skip_if_remote!(