openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 17:38:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden queued follow-up dispatch and listing

Browse Source
This commit is contained in:
Edward Frazer
2026-05-20 01:28:21 +00:00
parent 0d21de649b
commit 6402a8b290
10 changed files with 276 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+15 -9
View File
@@ -1943,15 +1943,11 @@
"description": "Optional JSON Schema used to constrain the final assistant message for this turn."
},
"permissions": {
"anyOf": [
{
"$ref": "#/definitions/PermissionProfileSelectionParams"
},
{
"type": "null"
}
],
"description": "Select a named permissions profile for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`. Use bounded `modifications` for supported turn adjustments instead of replacing the full permissions profile."
"description": "Select a named permissions profile id for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`.",
"type": [
"string",
"null"
]
},
"personality": {
"anyOf": [
@@ -1974,6 +1970,16 @@
"null"
]
},
"runtimeWorkspaceRoots": {
"description": "Replace the thread's runtime workspace roots for this turn and subsequent turns. Relative paths are resolved against the effective cwd for the turn.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
},
"sandboxPolicy": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+15 -68
View File
@@ -2492,65 +2492,6 @@
}
]
},
"PermissionProfileModificationParams": {
"oneOf": [
{
"description": "Additional concrete directory that should be writable.",
"properties": {
"path": {
"$ref": "#/definitions/AbsolutePathBuf"
},
"type": {
"enum": [
"additionalWritableRoot"
],
"title": "AdditionalWritableRootPermissionProfileModificationParamsType",
"type": "string"
}
},
"required": [
"path",
"type"
],
"title": "AdditionalWritableRootPermissionProfileModificationParams",
"type": "object"
}
]
},
"PermissionProfileSelectionParams": {
"oneOf": [
{
"description": "Select a named built-in or user-defined profile and optionally apply bounded modifications that Codex knows how to validate.",
"properties": {
"id": {
"type": "string"
},
"modifications": {
"items": {
"$ref": "#/definitions/PermissionProfileModificationParams"
},
"type": [
"array",
"null"
]
},
"type": {
"enum": [
"profile"
],
"title": "ProfilePermissionProfileSelectionParamsType",
"type": "string"
}
},
"required": [
"id",
"type"
],
"title": "ProfilePermissionProfileSelectionParams",
"type": "object"
}
]
},
"Personality": {
"enum": [
"none",
@@ -2789,15 +2730,11 @@
"description": "Optional JSON Schema used to constrain the final assistant message for this turn."
},
"permissions": {
"anyOf": [
{
"$ref": "#/definitions/PermissionProfileSelectionParams"
},
{
"type": "null"
}
],
"description": "Select a named permissions profile for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`. Use bounded `modifications` for supported turn adjustments instead of replacing the full permissions profile."
"description": "Select a named permissions profile id for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`.",
"type": [
"string",
"null"
]
},
"personality": {
"anyOf": [
@@ -2820,6 +2757,16 @@
"null"
]
},
"runtimeWorkspaceRoots": {
"description": "Replace the thread's runtime workspace roots for this turn and subsequent turns. Relative paths are resolved against the effective cwd for the turn.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
},
"sandboxPolicy": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+15 -9
View File
@@ -13159,15 +13159,11 @@
"description": "Optional JSON Schema used to constrain the final assistant message for this turn."
},
"permissions": {
"anyOf": [
{
"$ref": "#/definitions/v2/PermissionProfileSelectionParams"
},
{
"type": "null"
}
],
"description": "Select a named permissions profile for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`. Use bounded `modifications` for supported turn adjustments instead of replacing the full permissions profile."
"description": "Select a named permissions profile id for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`.",
"type": [
"string",
"null"
]
},
"personality": {
"anyOf": [
@@ -13190,6 +13186,16 @@
"null"
]
},
"runtimeWorkspaceRoots": {
"description": "Replace the thread's runtime workspace roots for this turn and subsequent turns. Relative paths are resolved against the effective cwd for the turn.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
},
"sandboxPolicy": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+15 -9
View File
@@ -9688,15 +9688,11 @@
"description": "Optional JSON Schema used to constrain the final assistant message for this turn."
},
"permissions": {
"anyOf": [
{
"$ref": "#/definitions/PermissionProfileSelectionParams"
},
{
"type": "null"
}
],
"description": "Select a named permissions profile for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`. Use bounded `modifications` for supported turn adjustments instead of replacing the full permissions profile."
"description": "Select a named permissions profile id for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`.",
"type": [
"string",
"null"
]
},
"personality": {
"anyOf": [
@@ -9719,6 +9715,16 @@
"null"
]
},
"runtimeWorkspaceRoots": {
"description": "Replace the thread's runtime workspace roots for this turn and subsequent turns. Relative paths are resolved against the effective cwd for the turn.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
},
"sandboxPolicy": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json
Generated
+44 -68
View File
@@ -236,6 +236,13 @@
],
"type": "object"
},
"ImageDetail": {
"enum": [
"high",
"original"
],
"type": "string"
},
"ModeKind": {
"description": "Initial collaboration mode to use when the TUI starts.",
"enum": [
@@ -258,65 +265,6 @@
],
"type": "string"
},
"PermissionProfileModificationParams": {
"oneOf": [
{
"description": "Additional concrete directory that should be writable.",
"properties": {
"path": {
"$ref": "#/definitions/AbsolutePathBuf"
},
"type": {
"enum": [
"additionalWritableRoot"
],
"title": "AdditionalWritableRootPermissionProfileModificationParamsType",
"type": "string"
}
},
"required": [
"path",
"type"
],
"title": "AdditionalWritableRootPermissionProfileModificationParams",
"type": "object"
}
]
},
"PermissionProfileSelectionParams": {
"oneOf": [
{
"description": "Select a named built-in or user-defined profile and optionally apply bounded modifications that Codex knows how to validate.",
"properties": {
"id": {
"type": "string"
},
"modifications": {
"items": {
"$ref": "#/definitions/PermissionProfileModificationParams"
},
"type": [
"array",
"null"
]
},
"type": {
"enum": [
"profile"
],
"title": "ProfilePermissionProfileSelectionParamsType",
"type": "string"
}
},
"required": [
"id",
"type"
],
"title": "ProfilePermissionProfileSelectionParams",
"type": "object"
}
]
},
"Personality": {
"enum": [
"none",
@@ -424,15 +372,11 @@
"description": "Optional JSON Schema used to constrain the final assistant message for this turn."
},
"permissions": {
"anyOf": [
{
"$ref": "#/definitions/PermissionProfileSelectionParams"
},
{
"type": "null"
}
],
"description": "Select a named permissions profile for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`. Use bounded `modifications` for supported turn adjustments instead of replacing the full permissions profile."
"description": "Select a named permissions profile id for this turn and subsequent turns. Cannot be combined with `sandboxPolicy`.",
"type": [
"string",
"null"
]
},
"personality": {
"anyOf": [
@@ -455,6 +399,16 @@
"null"
]
},
"runtimeWorkspaceRoots": {
"description": "Replace the thread's runtime workspace roots for this turn and subsequent turns. Relative paths are resolved against the effective cwd for the turn.",
"items": {
"type": "string"
},
"type": [
"array",
"null"
]
},
"sandboxPolicy": {
"anyOf": [
{
@@ -791,6 +745,17 @@
},
{
"properties": {
"detail": {
"anyOf": [
{
"$ref": "#/definitions/ImageDetail"
},
{
"type": "null"
}
],
"default": null
},
"type": {
"enum": [
"image"
@@ -811,6 +776,17 @@
},
{
"properties": {
"detail": {
"anyOf": [
{
"$ref": "#/definitions/ImageDetail"
},
{
"type": "null"
}
],
"default": null
},
"path": {
"type": "string"
},
codex-rs/app-server-protocol/src/protocol/v2/thread.rs
+6 -1
View File
@@ -714,13 +714,18 @@ pub struct ThreadQueueAddResponse {
#[ts(export_to = "v2/")]
pub struct ThreadQueueListParams {
pub thread_id: String,
#[ts(optional = nullable)]
pub cursor: Option<String>,
#[ts(optional = nullable)]
pub limit: Option<u32>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadQueueListResponse {
pub queued_turns: Vec<QueuedTurn>,
pub data: Vec<QueuedTurn>,
pub next_cursor: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
codex-rs/app-server/README.md
+1 -1
View File
@@ -148,7 +148,7 @@ Example with notification opt-out:
- `thread/goal/updated` — notification emitted whenever a thread goal changes; includes the full current goal.
- `thread/goal/cleared` — notification emitted whenever a thread goal is removed.
- `thread/queue/add` — experimental; persist a future `turn/start` request for a loaded thread. The queue stores the original `TurnStartParams`, dispatches the oldest pending row once the thread is idle, and emits `thread/queue/changed`.
- `thread/queue/list` — experimental; return the visible queued turns for a thread. Pending and failed rows are visible; the short-lived dispatch claim is internal.
- `thread/queue/list` — experimental; page through the visible queued turns for a thread with cursor/limit pagination. Pending and failed rows are visible; the short-lived dispatch claim is internal.
- `thread/queue/delete` — experimental; remove a visible queued turn by id.
- `thread/queue/reorder` — experimental; replace the visible queue order by queued-turn id.
- `thread/queue/changed` — experimental notification emitted after visible queue state changes, including restart recovery that surfaces an interrupted dispatch as failed.
codex-rs/app-server/src/request_processors/thread_queue_processor.rs
+42 -10
View File
@@ -1,5 +1,7 @@
use super::*;
const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25;
#[derive(Clone)]
pub(crate) struct ThreadQueueRequestProcessor {
thread_manager: Arc<ThreadManager>,
@@ -43,8 +45,15 @@ impl ThreadQueueRequestProcessor {
"`threadId` must match `turnStartParams.threadId`",
));
}
if self.thread_manager.get_thread(thread_id).await.is_err() {
return Err(invalid_request(format!("thread not found: {thread_id}")));
let thread = self
.thread_manager
.get_thread(thread_id)
.await
.map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?;
if thread.config_snapshot().await.ephemeral {
return Err(invalid_request(format!(
"ephemeral thread does not support queued turns: {thread_id}"
)));
}
TurnRequestProcessor::validate_v2_input_limit(&params.turn_start_params.input)?;
let payload = serde_json::to_vec(&params.turn_start_params).map_err(|err| {
@@ -74,8 +83,29 @@ impl ThreadQueueRequestProcessor {
params: ThreadQueueListParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let thread_id = parse_queue_thread_id(params.thread_id.as_str())?;
let queued_turns = self.list_visible_queued_turns(thread_id).await?;
Ok(Some(ThreadQueueListResponse { queued_turns }.into()))
let start = match params.cursor {
Some(cursor) => cursor
.parse::<usize>()
.map_err(|_| invalid_request(format!("invalid cursor: {cursor}")))?,
None => 0,
};
let limit = params
.limit
.unwrap_or(THREAD_QUEUE_LIST_DEFAULT_LIMIT as u32)
.max(1) as usize;
let records = self
.state_db()?
.list_visible_thread_queued_turns_page(thread_id, start, limit.saturating_add(1))
.await
.map_err(|err| internal_error(format!("failed to read queued turns: {err}")))?;
let has_next_page = records.len() > limit;
let data = records
.into_iter()
.take(limit)
.map(queued_turn_from_state)
.collect::<Result<Vec<_>, _>>()?;
let next_cursor = has_next_page.then(|| start.saturating_add(limit).to_string());
Ok(Some(ThreadQueueListResponse { data, next_cursor }.into()))
}
pub(crate) async fn thread_queue_delete(
@@ -248,12 +278,14 @@ impl ThreadQueueRequestProcessor {
}
}
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
if thread_state
.lock()
.await
.active_turn_snapshot()
.is_some_and(|turn| turn.id == turn_id)
{
let should_complete_dispatch = {
let thread_state = thread_state.lock().await;
thread_state
.active_turn_snapshot()
.is_some_and(|turn| turn.id == turn_id)
|| thread_state.last_terminal_turn_id.as_deref() == Some(turn_id.as_str())
};
if should_complete_dispatch {
self.complete_dispatch_after_turn_started(thread_id, turn_id.as_str())
.await;
}
codex-rs/app-server/tests/suite/v2/thread_queue.rs
+109 -9
View File
@@ -9,6 +9,7 @@ use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::QueuedTurnStatus;
use codex_app_server_protocol::RequestId;
@@ -30,10 +31,14 @@ use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
#[tokio::test]
async fn idle_queue_add_dispatches_serialized_turn_and_drains_visible_queue() -> Result<()> {
let responses = vec![create_final_assistant_message_sse_response("queued done")?];
let responses = vec![
create_final_assistant_message_sse_response("queued done")?,
create_final_assistant_message_sse_response("second queued done")?,
];
let server = create_mock_responses_server_sequence(responses).await;
let codex_home = TempDir::new()?;
write_queue_test_config(codex_home.path(), &server.uri(), "never")?;
@@ -95,6 +100,8 @@ async fn idle_queue_add_dispatches_serialized_turn_and_drains_visible_queue() ->
"thread/queue/list",
Some(serde_json::to_value(ThreadQueueListParams {
thread_id: thread.id.clone(),
cursor: None,
limit: None,
})?),
)
.await?;
@@ -103,18 +110,83 @@ async fn idle_queue_add_dispatches_serialized_turn_and_drains_visible_queue() ->
mcp.read_stream_until_response_message(RequestId::Integer(list_request_id)),
)
.await??;
let ThreadQueueListResponse { queued_turns } = to_response(list_response)?;
assert!(queued_turns.is_empty());
let ThreadQueueListResponse { data, next_cursor } = to_response(list_response)?;
assert!(data.is_empty());
assert_eq!(next_cursor, None);
queue_turn(&mut mcp, &thread.id, "second queued serialized input").await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
assert!(list_queue_ids(&mut mcp, &thread.id).await?.is_empty());
let requests = server
.received_requests()
.await
.expect("failed to fetch received requests");
assert_eq!(requests.len(), 1);
assert_eq!(requests.len(), 2);
assert!(
String::from_utf8_lossy(&requests[0].body).contains("queued serialized input"),
"queued turn payload should reach the model request after state round-trip"
);
assert!(
String::from_utf8_lossy(&requests[1].body).contains("second queued serialized input"),
"a later queued turn should still drain after a fast terminal dispatch"
);
Ok(())
}
#[tokio::test]
async fn queue_add_rejects_ephemeral_threads() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(vec![
create_final_assistant_message_sse_response("unused")?,
])
.await;
let codex_home = TempDir::new()?;
write_queue_test_config(codex_home.path(), &server.uri(), "never")?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
initialize_experimental(&mut mcp).await?;
let start_request_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
ephemeral: Some(true),
..Default::default()
})
.await?;
let start_response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(start_response)?;
let add_request_id = mcp
.send_raw_request(
"thread/queue/add",
Some(serde_json::to_value(ThreadQueueAddParams {
thread_id: thread.id.clone(),
turn_start_params: text_turn(&thread.id, "ephemeral queued turn"),
})?),
)
.await?;
let add_error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(add_request_id)),
)
.await??;
assert_eq!(add_error.error.code, INVALID_REQUEST_ERROR_CODE);
assert_eq!(
add_error.error.message,
format!(
"ephemeral thread does not support queued turns: {}",
thread.id
)
);
Ok(())
}
@@ -166,6 +238,26 @@ async fn busy_thread_queue_rows_support_list_reorder_and_delete_before_drain() -
list_queue_ids(&mut mcp, &thread.id).await?,
vec![first.clone(), second.clone()]
);
let first_page = list_queue_page(&mut mcp, &thread.id, None, Some(1)).await?;
assert_eq!(
first_page
.data
.into_iter()
.map(|queued_turn| queued_turn.id)
.collect::<Vec<_>>(),
vec![first.clone()]
);
let second_page =
list_queue_page(&mut mcp, &thread.id, first_page.next_cursor, Some(1)).await?;
assert_eq!(
second_page
.data
.into_iter()
.map(|queued_turn| queued_turn.id)
.collect::<Vec<_>>(),
vec![second.clone()]
);
assert_eq!(second_page.next_cursor, None);
let reorder_request_id = mcp
.send_raw_request(
@@ -335,11 +427,23 @@ async fn queue_turn(mcp: &mut McpProcess, thread_id: &str, text: &str) -> Result
}
async fn list_queue_ids(mcp: &mut McpProcess, thread_id: &str) -> Result<Vec<String>> {
let ThreadQueueListResponse { data, .. } = list_queue_page(mcp, thread_id, None, None).await?;
Ok(data.into_iter().map(|queued_turn| queued_turn.id).collect())
}
async fn list_queue_page(
mcp: &mut McpProcess,
thread_id: &str,
cursor: Option<String>,
limit: Option<u32>,
) -> Result<ThreadQueueListResponse> {
let request_id = mcp
.send_raw_request(
"thread/queue/list",
Some(serde_json::to_value(ThreadQueueListParams {
thread_id: thread_id.to_string(),
cursor,
limit,
})?),
)
.await?;
@@ -348,11 +452,7 @@ async fn list_queue_ids(mcp: &mut McpProcess, thread_id: &str) -> Result<Vec<Str
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let ThreadQueueListResponse { queued_turns } = to_response(response)?;
Ok(queued_turns
.into_iter()
.map(|queued_turn| queued_turn.id)
.collect())
to_response(response)
}
async fn delete_queue_turn(
codex-rs/state/src/runtime/queued_turns.rs
+14
View File
@@ -64,6 +64,16 @@ RETURNING
pub async fn list_visible_thread_queued_turns(
&self,
thread_id: ThreadId,
) -> anyhow::Result<Vec<crate::ThreadQueuedTurn>> {
self.list_visible_thread_queued_turns_page(thread_id, 0, i64::MAX as usize)
.await
}
pub async fn list_visible_thread_queued_turns_page(
&self,
thread_id: ThreadId,
offset: usize,
limit: usize,
) -> anyhow::Result<Vec<crate::ThreadQueuedTurn>> {
let rows = sqlx::query(
r#"
@@ -84,9 +94,13 @@ FROM thread_queued_turns
WHERE thread_id = ?
AND state IN ('pending', 'failed')
ORDER BY queue_order ASC
LIMIT ?
OFFSET ?
"#,
)
.bind(thread_id.to_string())
.bind(i64::try_from(limit)?)
.bind(i64::try_from(offset)?)
.fetch_all(self.pool.as_ref())
.await?;