diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index d8e126a7ed..0b27ff9356 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index b632df066b..11a2f4b97d 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index ea90d64cc6..8305c4e895 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 5f6023aeef..17ecb4d9ab 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json index 9dd1fd2f85..a02037eae3 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json @@ -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" }, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 59db1f1b6e..5e80b5c5c9 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -714,13 +714,18 @@ pub struct ThreadQueueAddResponse { #[ts(export_to = "v2/")] pub struct ThreadQueueListParams { pub thread_id: String, + #[ts(optional = nullable)] + pub cursor: Option, + #[ts(optional = nullable)] + pub limit: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadQueueListResponse { - pub queued_turns: Vec, + pub data: Vec, + pub next_cursor: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 17a5833faa..183bdcd797 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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. diff --git a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs index 28aa8bf0d8..f5f7000b19 100644 --- a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs @@ -1,5 +1,7 @@ use super::*; +const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25; + #[derive(Clone)] pub(crate) struct ThreadQueueRequestProcessor { thread_manager: Arc, @@ -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(¶ms.turn_start_params.input)?; let payload = serde_json::to_vec(¶ms.turn_start_params).map_err(|err| { @@ -74,8 +83,29 @@ impl ThreadQueueRequestProcessor { params: ThreadQueueListParams, ) -> Result, 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::() + .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::, _>>()?; + 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; } diff --git a/codex-rs/app-server/tests/suite/v2/thread_queue.rs b/codex-rs/app-server/tests/suite/v2/thread_queue.rs index 31179c23f9..4d037c7d31 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_queue.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_queue.rs @@ -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![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![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> { + 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, + limit: Option, +) -> Result { 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 anyhow::Result> { + 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> { 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?;