Add experimental thread queue APIs to app server (#38456)

## What changed

- Add experimental `thread/queue/add`, `list`, `update`, `delete`, `reorder`, and `start` requests for persistent queued user submissions.
- Automatically dispatch queued submissions in FIFO order after completed or failed turns, while leaving the queue paused after interruption.
- Emit `thread/queue/changed` notifications and expose paginated queue contents with stable submission and client message IDs.
- Export the new protocol types through the generated JSON Schema and TypeScript bindings.

## Testing

- Add app-server integration coverage for capability gating, CRUD, pagination, capacity limits, notifications, automatic dispatch, cold-thread resume, interruption, and explicit queue starts.

GitOrigin-RevId: 89caf5577192199bda77cf6c372a59d26cdd910e

codex-rs/app-server-protocol/schema/json/ServerNotification.json

@@ -4884,6 +4884,17 @@
       ],
       "type": "object"
     },
+    "ThreadQueueChangedNotification": {
+      "properties": {
+        "threadId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "threadId"
+      ],
+      "type": "object"
+    },
     "ThreadRealtimeAudioChunk": {
       "description": "EXPERIMENTAL - thread realtime audio chunk.",
       "properties": {
@@ -6260,6 +6271,26 @@
       "title": "Thread/goal/clearedNotification",
       "type": "object"
     },
+    {
+      "properties": {
+        "method": {
+          "enum": [
+            "thread/queue/changed"
+          ],
+          "title": "Thread/queue/changedNotificationMethod",
+          "type": "string"
+        },
+        "params": {
+          "$ref": "#/definitions/ThreadQueueChangedNotification"
+        }
+      },
+      "required": [
+        "method",
+        "params"
+      ],
+      "title": "Thread/queue/changedNotification",
+      "type": "object"
+    },
     {
       "properties": {
         "method": {

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -4577,6 +4577,26 @@
           "title": "Thread/goal/clearedNotification",
           "type": "object"
         },
+        {
+          "properties": {
+            "method": {
+              "enum": [
+                "thread/queue/changed"
+              ],
+              "title": "Thread/queue/changedNotificationMethod",
+              "type": "string"
+            },
+            "params": {
+              "$ref": "#/definitions/v2/ThreadQueueChangedNotification"
+            }
+          },
+          "required": [
+            "method",
+            "params"
+          ],
+          "title": "Thread/queue/changedNotification",
+          "type": "object"
+        },
         {
           "properties": {
             "method": {
@@ -15930,6 +15950,28 @@
         ],
         "type": "object"
       },
+      "QueuedSubmission": {
+        "properties": {
+          "clientUserMessageId": {
+            "type": "string"
+          },
+          "id": {
+            "type": "string"
+          },
+          "input": {
+            "items": {
+              "$ref": "#/definitions/v2/UserInput"
+            },
+            "type": "array"
+          }
+        },
+        "required": [
+          "clientUserMessageId",
+          "id",
+          "input"
+        ],
+        "type": "object"
+      },
       "RateLimitReachedType": {
         "enum": [
           "rate_limit_reached",
@@ -20504,6 +20546,19 @@
         "title": "ThreadNameUpdatedNotification",
         "type": "object"
       },
+      "ThreadQueueChangedNotification": {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "properties": {
+          "threadId": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "threadId"
+        ],
+        "title": "ThreadQueueChangedNotification",
+        "type": "object"
+      },
       "ThreadReadParams": {
         "$schema": "http://json-schema.org/draft-07/schema#",
         "properties": {

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json

@@ -12240,6 +12240,28 @@
       ],
       "type": "object"
     },
+    "QueuedSubmission": {
+      "properties": {
+        "clientUserMessageId": {
+          "type": "string"
+        },
+        "id": {
+          "type": "string"
+        },
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        }
+      },
+      "required": [
+        "clientUserMessageId",
+        "id",
+        "input"
+      ],
+      "type": "object"
+    },
     "RateLimitReachedType": {
       "enum": [
         "rate_limit_reached",
@@ -14555,6 +14577,26 @@
           "title": "Thread/goal/clearedNotification",
           "type": "object"
         },
+        {
+          "properties": {
+            "method": {
+              "enum": [
+                "thread/queue/changed"
+              ],
+              "title": "Thread/queue/changedNotificationMethod",
+              "type": "string"
+            },
+            "params": {
+              "$ref": "#/definitions/ThreadQueueChangedNotification"
+            }
+          },
+          "required": [
+            "method",
+            "params"
+          ],
+          "title": "Thread/queue/changedNotification",
+          "type": "object"
+        },
         {
           "properties": {
             "method": {
@@ -18256,6 +18298,19 @@
       "title": "ThreadNameUpdatedNotification",
       "type": "object"
     },
+    "ThreadQueueChangedNotification": {
+      "$schema": "http://json-schema.org/draft-07/schema#",
+      "properties": {
+        "threadId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "threadId"
+      ],
+      "title": "ThreadQueueChangedNotification",
+      "type": "object"
+    },
     "ThreadReadParams": {
       "$schema": "http://json-schema.org/draft-07/schema#",
       "properties": {

codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json

@@ -0,0 +1,13 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "threadId": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "threadId"
+  ],
+  "title": "ThreadQueueChangedNotification",
+  "type": "object"
+}

codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts

@@ -0,0 +1,6 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { UserInput } from "./UserInput";
+
+export type QueuedSubmission = { id: string, input: Array<UserInput>, clientUserMessageId: string, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ThreadQueueChangedNotification = { threadId: string, };

codex-rs/app-server-protocol/schema/typescript/v2/index.ts

@@ -362,6 +362,7 @@ export type { ProcessExitedNotification } from "./ProcessExitedNotification";
 export type { ProcessOutputDeltaNotification } from "./ProcessOutputDeltaNotification";
 export type { ProcessOutputStream } from "./ProcessOutputStream";
 export type { ProcessTerminalSize } from "./ProcessTerminalSize";
+export type { QueuedSubmission } from "./QueuedSubmission";
 export type { RateLimitReachedType } from "./RateLimitReachedType";
 export type { RateLimitResetCredit } from "./RateLimitResetCredit";
 export type { RateLimitResetCreditStatus } from "./RateLimitResetCreditStatus";
@@ -462,6 +463,7 @@ export type { ThreadMetadataGitInfoUpdateParams } from "./ThreadMetadataGitInfoU
 export type { ThreadMetadataUpdateParams } from "./ThreadMetadataUpdateParams";
 export type { ThreadMetadataUpdateResponse } from "./ThreadMetadataUpdateResponse";
 export type { ThreadNameUpdatedNotification } from "./ThreadNameUpdatedNotification";
+export type { ThreadQueueChangedNotification } from "./ThreadQueueChangedNotification";
 export type { ThreadReadParams } from "./ThreadReadParams";
 export type { ThreadReadResponse } from "./ThreadReadResponse";
 export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk";

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -574,6 +574,42 @@ client_request_definitions! {
         serialization: thread_id(params.thread_id),
         response: v2::ThreadGoalClearResponse,
     },
+    #[experimental("thread/queue/add")]
+    ThreadQueueAdd => "thread/queue/add" {
+        params: v2::ThreadQueueAddParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueAddResponse,
+    },
+    #[experimental("thread/queue/list")]
+    ThreadQueueList => "thread/queue/list" {
+        params: v2::ThreadQueueListParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueListResponse,
+    },
+    #[experimental("thread/queue/update")]
+    ThreadQueueUpdate => "thread/queue/update" {
+        params: v2::ThreadQueueUpdateParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueUpdateResponse,
+    },
+    #[experimental("thread/queue/delete")]
+    ThreadQueueDelete => "thread/queue/delete" {
+        params: v2::ThreadQueueDeleteParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueDeleteResponse,
+    },
+    #[experimental("thread/queue/reorder")]
+    ThreadQueueReorder => "thread/queue/reorder" {
+        params: v2::ThreadQueueReorderParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueReorderResponse,
+    },
+    #[experimental("thread/queue/start")]
+    ThreadQueueStart => "thread/queue/start" {
+        params: v2::ThreadQueueStartParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadQueueStartResponse,
+    },
     ThreadMetadataUpdate => "thread/metadata/update" {
         params: v2::ThreadMetadataUpdateParams,
         serialization: thread_id(params.thread_id),
@@ -1723,6 +1759,8 @@ server_notification_definitions! {
     ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification),
     ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification),
     ThreadGoalCleared => "thread/goal/cleared" (v2::ThreadGoalClearedNotification),
+    #[experimental("thread/queue/changed")]
+    ThreadQueueChanged => "thread/queue/changed" (v2::ThreadQueueChangedNotification),
     #[experimental("thread/environment/connected")]
     EnvironmentConnected => "thread/environment/connected" (v2::EnvironmentConnectionNotification),
     #[experimental("thread/environment/disconnected")]

codex-rs/app-server-protocol/src/protocol/v2/thread.rs

@@ -12,6 +12,7 @@ use super::ThreadSource;
 use super::Turn;
 use super::TurnEnvironmentParams;
 use super::TurnItemsView;
+use super::UserInput;
 use super::shared::v2_enum_from_core;
 use crate::JsonSchema;
 use crate::TS;
@@ -857,6 +858,113 @@ pub struct ThreadGoalClearResponse {
     pub cleared: bool,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct QueuedSubmission {
+    pub id: String,
+    pub input: Vec<UserInput>,
+    pub client_user_message_id: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueAddParams {
+    pub thread_id: String,
+    pub input: Vec<UserInput>,
+    pub client_user_message_id: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueAddResponse {
+    pub queued_submission: QueuedSubmission,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueListParams {
+    pub thread_id: String,
+    /// Opaque pagination cursor returned by a previous call.
+    #[ts(optional = nullable)]
+    pub cursor: Option<String>,
+    /// Optional page size; defaults to the standard thread-list page size.
+    #[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 data: Vec<QueuedSubmission>,
+    /// Opaque cursor for the next page, or `null` when no submissions remain.
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueUpdateParams {
+    pub thread_id: String,
+    pub queued_submission_id: String,
+    pub input: Vec<UserInput>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueUpdateResponse {
+    pub queued_submission: QueuedSubmission,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueDeleteParams {
+    pub thread_id: String,
+    pub queued_submission_id: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueDeleteResponse {
+    pub deleted: bool,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueReorderParams {
+    pub thread_id: String,
+    pub queued_submission_ids: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueReorderResponse {}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueStartParams {
+    pub thread_id: String,
+    #[ts(optional = nullable)]
+    pub queued_submission_id: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueStartResponse {
+    pub turn: Turn,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1768,6 +1876,13 @@ pub struct ThreadGoalClearedNotification {
     pub thread_id: String,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueChangedNotification {
+    pub thread_id: String,
+}
+
 /// Deprecated: Use `ContextCompaction` item type instead.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]

codex-rs/app-server/README.md

@@ -185,6 +185,13 @@ Example with notification opt-out:
 - `thread/goal/clear` — clear the current persisted goal for a materialized thread; returns whether a goal was removed and emits `thread/goal/cleared` when state changes.
 - `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 user turn for automatic FIFO submission when the thread next becomes idle.
+- `thread/queue/list` — experimental; return one page of a thread's queued turns.
+- `thread/queue/update` — experimental; edit a queued turn while preserving its stable submission ID, client message ID, and position.
+- `thread/queue/delete` — experimental; remove a queued turn by submission ID.
+- `thread/queue/reorder` — experimental; replace the order of a thread's queued turns.
+- `thread/queue/start` — experimental; start the queue head or a selected queued submission when the thread is idle.
+- `thread/queue/changed` — experimental notification emitted with the changed `threadId`.
 - `thread/settings/updated` — experimental notification emitted to subscribed clients when a loaded thread’s effective next-turn settings change; includes `threadId` and the full `threadSettings`.
 - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
 - `thread/archive` — move a thread’s rollout file into the archived directory and attempt to move any spawned descendant thread rollout files; returns `{}` on success and emits `thread/archived` for each archived thread.
@@ -776,6 +783,30 @@ Use `thread/goal/clear` to remove the current goal.
 { "method": "thread/goal/cleared", "params": { "threadId": "thr_123" } }
 ```
 
+### Example: Queue a follow-up user turn (experimental)
+
+Queued turns require `capabilities.experimentalApi = true`. Use `thread/queue/add` to persist a follow-up while a turn is running. Each thread can queue up to 100 messages, and the server starts the next queued turn when the thread becomes idle.
+
+A queued submission contains its user input and a required, client-provided `clientUserMessageId`. The server assigns a separate stable submission ID and preserves both IDs when the submission is edited. Application context and Responses API client metadata remain available on ordinary `turn/start`; queued submissions do not persist or replay those optional turn features.
+
+```json
+{ "method": "thread/queue/add", "id": 40, "params": {
+    "threadId": "thr_123",
+    "input": [{ "type": "text", "text": "Now fix the failing tests." }],
+    "clientUserMessageId": "019faba0-0000-7000-8000-000000000003"
+} }
+{ "id": 40, "result": { "queuedSubmission": {
+    "id": "019faba0-0000-7000-8000-000000000001",
+    "input": [{ "type": "text", "text": "Now fix the failing tests." }],
+    "clientUserMessageId": "019faba0-0000-7000-8000-000000000003"
+} } }
+{ "method": "thread/queue/changed", "params": { "threadId": "thr_123" } }
+```
+
+Use `thread/queue/list` to read the ordered queue. Pass optional `cursor` and `limit` values to request a page, and continue with the returned `nextCursor` until it is `null`. Each `thread/queue/changed` notification contains the changed `threadId`; fetch the current pages to refresh the queue. Update a queued turn by passing its `queuedSubmissionId` and replacement `input` to `thread/queue/update`; the submission keeps its IDs and position. Pass that ID to `thread/queue/delete` to remove it, or pass every queued ID in its new order as `queuedSubmissionIds` to `thread/queue/reorder`.
+
+Completed and failed turns automatically start the next queued submission. Interrupted turns leave the queue paused, including after `thread/resume`. Start the queue head with `thread/queue/start`, or select a queued submission by passing `queuedSubmissionId`. An idle thread starts a new turn and returns it; an active thread returns an invalid-request error and leaves the queue unchanged. The queued submission's client message ID remains stable, and its queue entry is removed when Core accepts the new turn. An ordinary `turn/start` does not consume queued submissions.
+
 ### Example: Archive a thread
 
 Use `thread/archive` to move the persisted rollout (stored as a JSONL file on disk) into the archived sessions directory and attempt to move any spawned descendant thread rollouts.

codex-rs/app-server/src/extensions.rs

@@ -6,6 +6,7 @@ use codex_analytics::AnalyticsEventsClient;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ThreadGoal;
 use codex_app_server_protocol::ThreadGoalUpdatedNotification;
+use codex_app_server_protocol::ThreadQueueChangedNotification;
 use codex_app_server_protocol::WarningNotification;
 use codex_core::NewThread;
 use codex_core::StartThreadOptions;
@@ -26,8 +27,8 @@ use codex_protocol::ThreadId;
 use codex_protocol::error::CodexErr;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
+use codex_queue_extension::QueuedItemService;
 use codex_rollout::state_db::StateDbHandle;
-use codex_thread_store::QueueStore;
 
 use crate::outgoing_message::OutgoingMessageSender;
 use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
@@ -45,8 +46,8 @@ pub(crate) struct ThreadExtensionDependencies {
     pub(crate) executor_skill_provider: Arc<dyn codex_skills_extension::SkillProvider>,
     pub(crate) git_attribution_base_url: String,
     pub(crate) http_client_factory: HttpClientFactory,
-    /// Process-scoped persistence backend for queued user messages.
-    pub(crate) queue_store: Option<Arc<dyn QueueStore>>,
+    /// Process-scoped queue shared by idle dispatch and app-server requests.
+    pub(crate) queue_service: Option<Arc<QueuedItemService>>,
 }
 
 pub(crate) fn thread_extensions<S>(
@@ -67,15 +68,10 @@ where
         executor_skill_provider,
         git_attribution_base_url,
         http_client_factory,
-        queue_store,
+        queue_service,
     } = dependencies;
     let mut builder = ExtensionRegistryBuilder::<Config>::with_event_sink(Arc::clone(&event_sink));
-    if let Some(queue) = queue_store {
-        let queue_service = Arc::new(codex_queue_extension::QueuedItemService::new(
-            queue,
-            thread_manager.clone(),
-            event_sink,
-        ));
+    if let Some(queue_service) = queue_service {
         codex_queue_extension::install(&mut builder, queue_service);
     }
     if let Some(state_db) = state_db {
@@ -171,6 +167,40 @@ const EXTENSION_WARNING_SUBSCRIBER_TIMEOUT: Duration = Duration::from_secs(10);
 impl ExtensionEventSink for AppServerExtensionEventSink {
     fn emit(&self, event: Event) {
         match event.msg {
+            EventMsg::ThreadQueueChanged(queue_event) => {
+                let thread_id = queue_event.thread_id;
+                if let Some(listener_command_tx) = self
+                    .thread_state_manager
+                    .current_listener_command_tx(thread_id)
+                {
+                    let command = ThreadListenerCommand::EmitThreadQueueChanged;
+                    if listener_command_tx.send(command).is_ok() {
+                        return;
+                    }
+                    tracing::warn!(
+                        "failed to enqueue extension queue update for {thread_id}: listener command channel is closed"
+                    );
+                }
+                let outgoing = Arc::clone(&self.outgoing);
+                let thread_state_manager = self.thread_state_manager.clone();
+                tokio::spawn(async move {
+                    let subscribed_connection_ids = thread_state_manager
+                        .subscribed_connection_ids(thread_id)
+                        .await;
+                    let outgoing = ThreadScopedOutgoingMessageSender::new(
+                        outgoing,
+                        subscribed_connection_ids,
+                        thread_id,
+                    );
+                    outgoing
+                        .send_server_notification(ServerNotification::ThreadQueueChanged(
+                            ThreadQueueChangedNotification {
+                                thread_id: thread_id.to_string(),
+                            },
+                        ))
+                        .await;
+                });
+            }
             EventMsg::ThreadGoalUpdated(thread_goal_event) => {
                 let thread_id = thread_goal_event.thread_id;
                 let turn_id = thread_goal_event.turn_id;

codex-rs/app-server/src/in_process.rs

@@ -108,6 +108,7 @@ fn server_notification_requires_delivery(notification: &ServerNotification) -> b
     matches!(
         notification,
         ServerNotification::TurnCompleted(_)
+            | ServerNotification::ThreadQueueChanged(_)
             | ServerNotification::ThreadSettingsUpdated(_)
             | ServerNotification::ExternalAgentConfigImportCompleted(_)
     )
@@ -780,6 +781,7 @@ mod tests {
     use codex_app_server_protocol::ConfigRequirementsReadResponse;
     use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
     use codex_app_server_protocol::SessionSource as ApiSessionSource;
+    use codex_app_server_protocol::ThreadQueueChangedNotification;
     use codex_app_server_protocol::ThreadStartParams;
     use codex_app_server_protocol::ThreadStartResponse;
     use codex_app_server_protocol::Turn;
@@ -996,6 +998,11 @@ mod tests {
                 },
             })
         ));
+        assert!(server_notification_requires_delivery(
+            &ServerNotification::ThreadQueueChanged(ThreadQueueChangedNotification {
+                thread_id: "thread-1".to_string(),
+            })
+        ));
         assert!(server_notification_requires_delivery(
             &ServerNotification::ExternalAgentConfigImportCompleted(
                 ExternalAgentConfigImportCompletedNotification {

codex-rs/app-server/src/mcp_refresh.rs

@@ -339,7 +339,7 @@ enabled = false
                         executor_skill_provider: Arc::clone(&executor_skill_provider),
                         git_attribution_base_url: good_config.chatgpt_base_url.clone(),
                         http_client_factory: good_config.http_client_factory(),
-                        queue_store: None,
+                        queue_service: None,
                     },
                 ),
                 Arc::new(CodexHomeUserInstructionsProvider::new(

codex-rs/app-server/src/message_processor.rs

@@ -37,6 +37,7 @@ use crate::request_processors::ProcessExecRequestProcessor;
 use crate::request_processors::RemoteControlRequestProcessor;
 use crate::request_processors::SearchRequestProcessor;
 use crate::request_processors::ThreadGoalRequestProcessor;
+use crate::request_processors::ThreadQueueRequestProcessor;
 use crate::request_processors::ThreadRequestProcessor;
 use crate::request_processors::TurnRequestProcessor;
 use crate::request_processors::WindowsSandboxRequestProcessor;
@@ -77,6 +78,7 @@ use codex_protocol::ThreadId;
 use codex_protocol::mcp::ClientMcpExtensions;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::W3cTraceContext;
+use codex_queue_extension::QueuedItemService;
 use codex_rollout::StateDbHandle;
 use codex_state::log_db::LogDbLayer;
 use codex_thread_store::LocalQueueStore;
@@ -121,6 +123,7 @@ pub(crate) struct MessageProcessor {
     remote_control_processor: RemoteControlRequestProcessor,
     search_processor: SearchRequestProcessor,
     thread_goal_processor: ThreadGoalRequestProcessor,
+    thread_queue_processor: ThreadQueueRequestProcessor,
     thread_processor: ThreadRequestProcessor,
     turn_processor: TurnRequestProcessor,
     windows_sandbox_processor: WindowsSandboxRequestProcessor,
@@ -269,7 +272,17 @@ impl MessageProcessor {
             ),
         );
         let goal_service = Arc::new(GoalService::new());
+        let extension_event_sink =
+            app_server_extension_event_sink(outgoing.clone(), thread_state_manager.clone());
+        let mut queue_service = None;
         let thread_manager = Arc::new_cyclic(|thread_manager| {
+            queue_service = queue_store.map(|queue| {
+                Arc::new(QueuedItemService::new(
+                    queue,
+                    thread_manager.clone(),
+                    Arc::clone(&extension_event_sink),
+                ))
+            });
             let manager = ThreadManager::new(
                 config.as_ref(),
                 auth_manager.clone(),
@@ -280,10 +293,7 @@ impl MessageProcessor {
                 thread_extensions(
                     guardian_agent_spawner(thread_manager.clone()),
                     ThreadExtensionDependencies {
-                        event_sink: app_server_extension_event_sink(
-                            outgoing.clone(),
-                            thread_state_manager.clone(),
-                        ),
+                        event_sink: Arc::clone(&extension_event_sink),
                         auth_manager: auth_manager.clone(),
                         state_db: state_db.clone(),
                         analytics_events_client: analytics_events_client.clone(),
@@ -293,7 +303,7 @@ impl MessageProcessor {
                         executor_skill_provider: Arc::clone(&executor_skill_provider),
                         git_attribution_base_url: config.chatgpt_base_url.clone(),
                         http_client_factory: config.http_client_factory(),
-                        queue_store,
+                        queue_service: queue_service.clone(),
                     },
                 ),
                 Arc::new(CodexHomeUserInstructionsProvider::new(
@@ -432,6 +442,12 @@ impl MessageProcessor {
             state_db.clone(),
             Arc::clone(&goal_service),
         );
+        let thread_queue_processor = ThreadQueueRequestProcessor::new(
+            Arc::clone(&thread_manager),
+            Arc::clone(&thread_store),
+            outgoing.clone(),
+            queue_service,
+        );
         let thread_processor = ThreadRequestProcessor::new(
             auth_manager.clone(),
             Arc::clone(&thread_manager),
@@ -522,6 +538,7 @@ impl MessageProcessor {
             remote_control_processor,
             search_processor,
             thread_goal_processor,
+            thread_queue_processor,
             thread_processor,
             turn_processor,
             windows_sandbox_processor,
@@ -1124,6 +1141,36 @@ impl MessageProcessor {
                     .thread_goal_clear(request_id.clone(), params)
                     .await
             }
+            ClientRequest::ThreadQueueAdd { params, .. } => self
+                .thread_queue_processor
+                .add(params)
+                .await
+                .map(|response| Some(response.into())),
+            ClientRequest::ThreadQueueList { params, .. } => self
+                .thread_queue_processor
+                .list(params)
+                .await
+                .map(|response| Some(response.into())),
+            ClientRequest::ThreadQueueUpdate { params, .. } => self
+                .thread_queue_processor
+                .update(params)
+                .await
+                .map(|response| Some(response.into())),
+            ClientRequest::ThreadQueueDelete { params, .. } => self
+                .thread_queue_processor
+                .delete(params)
+                .await
+                .map(|response| Some(response.into())),
+            ClientRequest::ThreadQueueReorder { params, .. } => self
+                .thread_queue_processor
+                .reorder(params)
+                .await
+                .map(|response| Some(response.into())),
+            ClientRequest::ThreadQueueStart { params, .. } => self
+                .thread_queue_processor
+                .start(&request_id, params)
+                .await
+                .map(|response| Some(response.into())),
             ClientRequest::ThreadMetadataUpdate { params, .. } => {
                 self.thread_processor.thread_metadata_update(params).await
             }

codex-rs/app-server/src/request_processors.rs

@@ -548,6 +548,7 @@ mod search;
 mod thread_enrichment;
 mod thread_fork_goal;
 mod thread_processor;
+mod thread_queue_processor;
 mod thread_sections;
 mod token_usage_replay;
 mod turn_processor;
@@ -572,6 +573,7 @@ pub(crate) use remote_control_processor::RemoteControlRequestProcessor;
 pub(crate) use search::SearchRequestProcessor;
 pub(crate) use thread_goal_processor::ThreadGoalRequestProcessor;
 pub(crate) use thread_processor::ThreadRequestProcessor;
+pub(crate) use thread_queue_processor::ThreadQueueRequestProcessor;
 pub(crate) use turn_processor::TurnRequestProcessor;
 pub(crate) use windows_sandbox_processor::WindowsSandboxRequestProcessor;
 

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -1,5 +1,6 @@
 use super::*;
 use crate::extensions::send_thread_warning;
+use codex_app_server_protocol::ThreadQueueChangedNotification;
 use codex_extension_api::ThreadIdleCause;
 use codex_protocol::config_types::MultiAgentMode;
 
@@ -501,6 +502,23 @@ pub(super) async fn handle_thread_listener_command(
                 ))
                 .await;
         }
+        ThreadListenerCommand::EmitThreadQueueChanged => {
+            let subscribed_connection_ids = thread_state_manager
+                .subscribed_connection_ids(conversation_id)
+                .await;
+            let outgoing = ThreadScopedOutgoingMessageSender::new(
+                Arc::clone(outgoing),
+                subscribed_connection_ids,
+                conversation_id,
+            );
+            outgoing
+                .send_server_notification(ServerNotification::ThreadQueueChanged(
+                    ThreadQueueChangedNotification {
+                        thread_id: conversation_id.to_string(),
+                    },
+                ))
+                .await;
+        }
         ThreadListenerCommand::EmitWarning { message } => {
             send_thread_warning(outgoing, thread_state_manager, conversation_id, message).await;
         }

codex-rs/app-server/src/request_processors/thread_queue_processor.rs

@@ -0,0 +1,336 @@
+use std::sync::Arc;
+
+use crate::error_code::internal_error;
+use crate::error_code::invalid_request;
+use crate::outgoing_message::ConnectionRequestId;
+use crate::outgoing_message::OutgoingMessageSender;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::QueuedSubmission;
+use codex_app_server_protocol::ThreadQueueAddParams;
+use codex_app_server_protocol::ThreadQueueAddResponse;
+use codex_app_server_protocol::ThreadQueueDeleteParams;
+use codex_app_server_protocol::ThreadQueueDeleteResponse;
+use codex_app_server_protocol::ThreadQueueListParams;
+use codex_app_server_protocol::ThreadQueueListResponse;
+use codex_app_server_protocol::ThreadQueueReorderParams;
+use codex_app_server_protocol::ThreadQueueReorderResponse;
+use codex_app_server_protocol::ThreadQueueStartParams;
+use codex_app_server_protocol::ThreadQueueStartResponse;
+use codex_app_server_protocol::ThreadQueueUpdateParams;
+use codex_app_server_protocol::ThreadQueueUpdateResponse;
+use codex_app_server_protocol::Turn;
+use codex_app_server_protocol::TurnItemsView;
+use codex_app_server_protocol::TurnStatus;
+use codex_app_server_protocol::UserInput;
+use codex_core::CodexThread;
+use codex_core::NotSubmittedReason;
+use codex_core::StartIfIdleSubmission;
+use codex_core::ThreadManager;
+use codex_core::TurnInput;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::SubAgentSource;
+use codex_queue_extension::QueueServiceError;
+use codex_queue_extension::QueuedItem;
+use codex_queue_extension::QueuedItemService;
+use codex_thread_store::ReadThreadParams;
+use codex_thread_store::ThreadStore;
+use codex_thread_store::ThreadStoreError;
+
+use super::TurnRequestProcessor;
+use super::thread_processor::THREAD_LIST_DEFAULT_LIMIT;
+use super::thread_processor::THREAD_LIST_MAX_LIMIT;
+use super::turn_processor::DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR;
+use super::turn_processor::can_accept_direct_input;
+use super::turn_processor::validate_user_input_image_urls;
+
+const DIRECT_INPUT_TO_UNLOADED_SUBAGENT_ERROR: &str =
+    "direct app-server input is not allowed for unloaded spawned sub-agents";
+
+pub(crate) struct ThreadQueueRequestProcessor {
+    thread_manager: Arc<ThreadManager>,
+    thread_store: Arc<dyn ThreadStore>,
+    outgoing: Arc<OutgoingMessageSender>,
+    service: Option<Arc<QueuedItemService>>,
+}
+
+impl ThreadQueueRequestProcessor {
+    pub(crate) fn new(
+        thread_manager: Arc<ThreadManager>,
+        thread_store: Arc<dyn ThreadStore>,
+        outgoing: Arc<OutgoingMessageSender>,
+        service: Option<Arc<QueuedItemService>>,
+    ) -> Self {
+        Self {
+            thread_manager,
+            thread_store,
+            outgoing,
+            service,
+        }
+    }
+
+    pub(crate) async fn add(
+        &self,
+        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?;
+        ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
+        let queued_item = self
+            .service()?
+            .enqueue(
+                thread_id,
+                submission_into_turn_input(params.input, Some(params.client_user_message_id)),
+            )
+            .await
+            .map_err(queue_error)?;
+        Ok(ThreadQueueAddResponse {
+            queued_submission: api_queued_submission(queued_item)?,
+        })
+    }
+
+    pub(crate) async fn list(
+        &self,
+        params: ThreadQueueListParams,
+    ) -> Result<ThreadQueueListResponse, JSONRPCErrorError> {
+        let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
+        let offset = params
+            .cursor
+            .as_deref()
+            .map(str::parse::<usize>)
+            .transpose()
+            .map_err(|error| invalid_request(format!("invalid queue pagination cursor: {error}")))?
+            .unwrap_or_default();
+        let limit = params
+            .limit
+            .map(|value| value as usize)
+            .unwrap_or(THREAD_LIST_DEFAULT_LIMIT)
+            .clamp(1, THREAD_LIST_MAX_LIMIT);
+        let mut items = self
+            .service()?
+            .list_page(thread_id, offset, limit.saturating_add(1))
+            .await
+            .map_err(queue_error)?;
+        let next_cursor = if items.len() > limit {
+            items.truncate(limit);
+            Some(offset.saturating_add(limit).to_string())
+        } else {
+            None
+        };
+        Ok(ThreadQueueListResponse {
+            data: items
+                .into_iter()
+                .map(api_queued_submission)
+                .collect::<Result<Vec<_>, _>>()?,
+            next_cursor,
+        })
+    }
+
+    pub(crate) async fn update(
+        &self,
+        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?;
+        ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
+        let queued_item = self
+            .service()?
+            .update(
+                thread_id,
+                params.queued_submission_id.clone(),
+                submission_into_turn_input(params.input, /*client_user_message_id*/ None),
+            )
+            .await
+            .map_err(queue_error)?
+            .ok_or_else(|| {
+                invalid_request(format!(
+                    "queued submission not found: {}",
+                    params.queued_submission_id
+                ))
+            })?;
+        Ok(ThreadQueueUpdateResponse {
+            queued_submission: api_queued_submission(queued_item)?,
+        })
+    }
+
+    pub(crate) async fn delete(
+        &self,
+        params: ThreadQueueDeleteParams,
+    ) -> Result<ThreadQueueDeleteResponse, JSONRPCErrorError> {
+        let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
+        let deleted = self
+            .service()?
+            .delete(thread_id, params.queued_submission_id)
+            .await
+            .map_err(queue_error)?;
+        Ok(ThreadQueueDeleteResponse { deleted })
+    }
+
+    pub(crate) async fn reorder(
+        &self,
+        params: ThreadQueueReorderParams,
+    ) -> Result<ThreadQueueReorderResponse, JSONRPCErrorError> {
+        let (thread_id, _, _) = self.require_thread(&params.thread_id).await?;
+        self.service()?
+            .reorder(thread_id, params.queued_submission_ids)
+            .await
+            .map_err(queue_error)?;
+        Ok(ThreadQueueReorderResponse {})
+    }
+
+    pub(crate) async fn start(
+        &self,
+        request_id: &ConnectionRequestId,
+        params: ThreadQueueStartParams,
+    ) -> Result<ThreadQueueStartResponse, JSONRPCErrorError> {
+        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"))?;
+        let submission = self
+            .service()?
+            .start(
+                thread.as_ref(),
+                params.queued_submission_id,
+                self.outgoing.request_trace_context(request_id).await,
+            )
+            .await
+            .map_err(queue_error)?;
+        let turn_id = match submission {
+            StartIfIdleSubmission::Started { turn_id } => turn_id,
+            StartIfIdleSubmission::NotSubmitted {
+                reason: NotSubmittedReason::NotIdle | NotSubmittedReason::PendingTriggerTurn,
+            } => {
+                return Err(invalid_request(
+                    "thread already has an active or pending turn",
+                ));
+            }
+            StartIfIdleSubmission::NotSubmitted { reason } => {
+                return Err(internal_error(format!(
+                    "Core declined to start queued user message: {reason:?}"
+                )));
+            }
+        };
+        self.outgoing
+            .record_request_turn_id(request_id, &turn_id)
+            .await;
+        Ok(ThreadQueueStartResponse {
+            turn: Turn {
+                id: turn_id,
+                items: vec![],
+                items_view: TurnItemsView::NotLoaded,
+                error: None,
+                status: TurnStatus::InProgress,
+                started_at: None,
+                completed_at: None,
+                duration_ms: None,
+            },
+        })
+    }
+
+    fn service(&self) -> Result<&QueuedItemService, JSONRPCErrorError> {
+        self.service
+            .as_deref()
+            .ok_or_else(|| invalid_request("user message queue is unavailable"))
+    }
+
+    async fn require_thread(
+        &self,
+        raw_thread_id: &str,
+    ) -> Result<(ThreadId, Option<Arc<CodexThread>>, SessionSource), 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) =
+            self.thread_manager.get_thread(thread_id).await
+        {
+            let snapshot = thread.config_snapshot().await;
+            if snapshot.ephemeral {
+                return Err(invalid_request(format!(
+                    "ephemeral thread does not support queued submissions: {thread_id}"
+                )));
+            }
+            (Some(thread), snapshot.session_source)
+        } else {
+            let stored = self
+                .thread_store
+                .read_thread(ReadThreadParams {
+                    thread_id,
+                    include_archived: true,
+                    include_history: false,
+                })
+                .await
+                .map_err(|error| match error {
+                    ThreadStoreError::ThreadNotFound { .. } => {
+                        invalid_request(format!("thread not found: {thread_id}"))
+                    }
+                    error => internal_error(format!("failed to read thread: {error}")),
+                })?;
+            if stored.archived_at.is_some() {
+                return Err(invalid_request(format!(
+                    "session {thread_id} is archived. Run `codex unarchive {thread_id}` to unarchive it first."
+                )));
+            }
+            (None, stored.source)
+        };
+
+        Ok((thread_id, loaded_thread, source))
+    }
+}
+
+fn ensure_direct_input_allowed(
+    loaded_thread: Option<&CodexThread>,
+    source: &SessionSource,
+) -> Result<(), JSONRPCErrorError> {
+    match loaded_thread {
+        Some(thread) if !can_accept_direct_input(thread.multi_agent_version(), source) => Err(
+            invalid_request(DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR),
+        ),
+        None if matches!(
+            source,
+            SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. })
+        ) =>
+        {
+            Err(invalid_request(DIRECT_INPUT_TO_UNLOADED_SUBAGENT_ERROR))
+        }
+        _ => Ok(()),
+    }
+}
+
+fn submission_into_turn_input(
+    input: Vec<UserInput>,
+    client_user_message_id: Option<String>,
+) -> TurnInput {
+    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 {
+    match error {
+        QueueServiceError::InputTooLarge { actual_chars } => {
+            TurnRequestProcessor::input_too_large_error(actual_chars)
+        }
+        error @ (QueueServiceError::InvalidInput | QueueServiceError::InvalidAttachment(_)) => {
+            invalid_request(error.to_string())
+        }
+        QueueServiceError::Storage(ThreadStoreError::InvalidRequest { message }) => {
+            invalid_request(message)
+        }
+        error => internal_error(format!("queued submission operation failed: {error}")),
+    }
+}
+
+fn api_queued_submission(value: QueuedItem) -> Result<QueuedSubmission, JSONRPCErrorError> {
+    let TurnInput::UserInput { content, client_id } = value.input else {
+        return Err(internal_error(
+            "queued submission does not contain user input",
+        ));
+    };
+    Ok(QueuedSubmission {
+        id: value.id,
+        input: content.into_iter().map(Into::into).collect(),
+        client_user_message_id: client_id
+            .ok_or_else(|| internal_error("queued submission is missing its client message id"))?,
+    })
+}

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -16,7 +16,7 @@ use codex_skills::system_cache_root_dir;
 use crate::image_url::REMOTE_IMAGE_URL_ERROR;
 use crate::image_url::is_remote_image_url;
 
-const DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR: &str =
+pub(super) const DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR: &str =
     "direct app-server input is not allowed for multi-agent v2 sub-agents";
 
 /// Mirrors the direct-input policy in both request validation and thread capability responses.
@@ -31,7 +31,9 @@ pub(super) fn can_accept_direct_input(
         )
 }
 
-fn validate_user_input_image_urls(input: &[V2UserInput]) -> Result<(), JSONRPCErrorError> {
+pub(super) fn validate_user_input_image_urls(
+    input: &[V2UserInput],
+) -> Result<(), JSONRPCErrorError> {
     if input.iter().any(|item| {
         matches!(
             item,
@@ -451,7 +453,7 @@ impl TurnRequestProcessor {
             .await
     }
 
-    fn input_too_large_error(actual_chars: usize) -> JSONRPCErrorError {
+    pub(super) fn input_too_large_error(actual_chars: usize) -> JSONRPCErrorError {
         let mut error = invalid_params(format!(
             "Input exceeds the maximum length of {MAX_USER_INPUT_TEXT_CHARS} characters."
         ));
@@ -463,7 +465,7 @@ impl TurnRequestProcessor {
         error
     }
 
-    fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> {
+    pub(super) fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> {
         let actual_chars: usize = items.iter().map(V2UserInput::text_char_count).sum();
         if actual_chars > MAX_USER_INPUT_TEXT_CHARS {
             return Err(Self::input_too_large_error(actual_chars));

codex-rs/app-server/src/thread_state.rs

@@ -61,6 +61,8 @@ pub(crate) enum ThreadListenerCommand {
         turn_id: Option<String>,
         goal: ThreadGoal,
     },
+    // EmitThreadQueueChanged orders durable queue updates with thread notifications.
+    EmitThreadQueueChanged,
     // EmitWarning is used to order extension warnings with other thread notifications.
     EmitWarning {
         message: String,

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -90,6 +90,7 @@ mod thread_loaded_list;
 mod thread_memory_mode_set;
 mod thread_metadata_update;
 mod thread_name_websocket;
+mod thread_queue;
 mod thread_read;
 mod thread_resume;
 mod thread_revert;

codex-rs/app-server/tests/suite/v2/thread_queue.rs

@@ -0,0 +1,978 @@
+use std::time::Duration;
+
+use anyhow::Context;
+use anyhow::Result;
+use app_test_support::MockResponsesConfig;
+use app_test_support::TestAppServer;
+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 app_test_support::create_shell_command_sse_response;
+use codex_app_server_protocol::ClientInfo;
+use codex_app_server_protocol::ClientRequest;
+use codex_app_server_protocol::CommandExecutionApprovalDecision;
+use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
+use codex_app_server_protocol::InitializeCapabilities;
+use codex_app_server_protocol::ItemStartedNotification;
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::QueuedSubmission;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ServerRequest;
+use codex_app_server_protocol::ThreadItem;
+use codex_app_server_protocol::ThreadQueueAddParams;
+use codex_app_server_protocol::ThreadQueueAddResponse;
+use codex_app_server_protocol::ThreadQueueChangedNotification;
+use codex_app_server_protocol::ThreadQueueDeleteParams;
+use codex_app_server_protocol::ThreadQueueDeleteResponse;
+use codex_app_server_protocol::ThreadQueueListParams;
+use codex_app_server_protocol::ThreadQueueListResponse;
+use codex_app_server_protocol::ThreadQueueReorderParams;
+use codex_app_server_protocol::ThreadQueueReorderResponse;
+use codex_app_server_protocol::ThreadQueueStartParams;
+use codex_app_server_protocol::ThreadQueueStartResponse;
+use codex_app_server_protocol::ThreadQueueUpdateParams;
+use codex_app_server_protocol::ThreadQueueUpdateResponse;
+use codex_app_server_protocol::ThreadResumeParams;
+use codex_app_server_protocol::ThreadResumeResponse;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnCompletedNotification;
+use codex_app_server_protocol::TurnInterruptParams;
+use codex_app_server_protocol::TurnInterruptResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::TurnStatus;
+use codex_app_server_protocol::UserInput;
+use core_test_support::skip_if_remote;
+use pretty_assertions::assert_eq;
+use serde_json::Value;
+use serde_json::json;
+use tempfile::TempDir;
+use tokio::time::timeout;
+use wiremock::MockServer;
+
+const READ_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 10);
+
+#[tokio::test]
+async fn queue_requires_experimental_handshake() -> Result<()> {
+    let (mut app, codex_home, _server) = queue_app(Vec::new()).await?;
+    let thread = app.start_thread(ThreadStartParams::default()).await?.thread;
+    let queue = list_queue(&mut app, &thread.id).await?;
+    assert!(queue.data.is_empty());
+    drop(app);
+
+    let mut app = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build()
+        .await?;
+    app.initialize_with_capabilities(
+        ClientInfo {
+            name: "queue-experimental-gate".to_string(),
+            title: None,
+            version: "0.1.0".to_string(),
+        },
+        Some(InitializeCapabilities::default()),
+    )
+    .await?;
+    let request_id = app
+        .send_raw_request("thread/start", Some(json!({})))
+        .await?;
+    let thread = timeout(
+        READ_TIMEOUT,
+        app.read_response::<ThreadStartResponse>(request_id),
+    )
+    .await??
+    .thread;
+    let request_id = app
+        .send_raw_request(
+            "thread/queue/list",
+            Some(serde_json::to_value(ThreadQueueListParams {
+                thread_id: thread.id,
+                cursor: None,
+                limit: None,
+            })?),
+        )
+        .await?;
+    let error: JSONRPCError = timeout(
+        READ_TIMEOUT,
+        app.read_stream_until_error_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    assert!(error.error.message.contains("experimental"));
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let responses = vec![
+        blocked_turn_response()?,
+        create_final_assistant_message_sse_response("active done")?,
+        create_final_assistant_message_sse_response("queued done")?,
+    ];
+    let (mut app, _codex_home, _server) = queue_app(responses).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let (_, approval_id) = start_blocked_turn(&mut app, &thread_id).await?;
+    let first = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            client_user_message_id: "first-client-message".to_string(),
+            ..submission(&thread_id, "first")
+        },
+    )
+    .await?;
+    let second = queue_item(&mut app, submission(&thread_id, "second")).await?;
+    let first_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    let second_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        first_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+    assert_eq!(
+        second_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+
+    let updated: ThreadQueueUpdateResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueUpdate {
+            request_id,
+            params: ThreadQueueUpdateParams {
+                thread_id: thread_id.clone(),
+                queued_submission_id: first.id.clone(),
+                input: vec![text("first edited")],
+            },
+        })
+        .await?;
+    assert_eq!(updated.queued_submission.id, first.id);
+    assert_eq!(
+        updated.queued_submission.client_user_message_id,
+        first.client_user_message_id
+    );
+    assert_eq!(updated.queued_submission.input, vec![text("first edited")]);
+    let update_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        update_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+
+    let invalid_reorder = app
+        .send_raw_request(
+            "thread/queue/reorder",
+            Some(serde_json::to_value(ThreadQueueReorderParams {
+                thread_id: thread_id.clone(),
+                queued_submission_ids: vec![first.id.clone()],
+            })?),
+        )
+        .await?;
+    let error: JSONRPCError = timeout(
+        READ_TIMEOUT,
+        app.read_stream_until_error_message(RequestId::Integer(invalid_reorder)),
+    )
+    .await??;
+    assert_eq!(
+        error.error.message,
+        "queue reorder must include every queued submission exactly once"
+    );
+
+    let _: ThreadQueueReorderResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueReorder {
+            request_id,
+            params: ThreadQueueReorderParams {
+                thread_id: thread_id.clone(),
+                queued_submission_ids: vec![second.id.clone(), first.id.clone()],
+            },
+        })
+        .await?;
+    let reorder_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        reorder_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+    let deleted: ThreadQueueDeleteResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueDelete {
+            request_id,
+            params: ThreadQueueDeleteParams {
+                thread_id: thread_id.clone(),
+                queued_submission_id: second.id,
+            },
+        })
+        .await?;
+    assert!(deleted.deleted);
+    let delete_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        delete_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+
+    decline_approval(&mut app, approval_id).await?;
+    for _ in 0..2 {
+        let _: TurnCompletedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+    }
+    let drain_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(drain_change, ThreadQueueChangedNotification { thread_id });
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_list_returns_ordered_pages_and_lightweight_notifications() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let (mut app, _codex_home, _server) = queue_app(vec![blocked_turn_response()?]).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let _blocked = start_blocked_turn(&mut app, &thread_id).await?;
+    let first = queue_item(
+        &mut app,
+        submission(&thread_id, "fitting queued submission"),
+    )
+    .await?;
+    let queued = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            thread_id: thread_id.clone(),
+            input: vec![text(&"x".repeat(64 * 1024))],
+            client_user_message_id: "oversized-snapshot".to_string(),
+        },
+    )
+    .await?;
+    let initial_change: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        initial_change,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+    let changed: ThreadQueueChangedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
+    assert_eq!(
+        changed,
+        ThreadQueueChangedNotification {
+            thread_id: thread_id.clone(),
+        }
+    );
+    let first_page: ThreadQueueListResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueList {
+            request_id,
+            params: ThreadQueueListParams {
+                thread_id: thread_id.clone(),
+                cursor: None,
+                limit: Some(1),
+            },
+        })
+        .await?;
+    assert_eq!(first_page.data, vec![first.clone()]);
+    assert_eq!(first_page.next_cursor, Some("1".to_string()));
+    let second_page: ThreadQueueListResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueList {
+            request_id,
+            params: ThreadQueueListParams {
+                thread_id: thread_id.clone(),
+                cursor: first_page.next_cursor,
+                limit: Some(1),
+            },
+        })
+        .await?;
+    assert_eq!(second_page.data, vec![queued.clone()]);
+    assert_eq!(second_page.next_cursor, None);
+    assert_eq!(
+        list_queue(&mut app, &thread_id).await?.data,
+        vec![first, queued]
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_rejects_messages_after_reaching_its_capacity() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let (mut app, _codex_home, _server) = queue_app(vec![blocked_turn_response()?]).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let _blocked = start_blocked_turn(&mut app, &thread_id).await?;
+
+    for index in 0..100 {
+        queue_item(&mut app, submission(&thread_id, &format!("queued {index}"))).await?;
+    }
+    assert_eq!(list_queue(&mut app, &thread_id).await?.data.len(), 100);
+
+    let request_id = app
+        .send_raw_request(
+            "thread/queue/add",
+            Some(serde_json::to_value(submission(
+                &thread_id,
+                "one too many",
+            ))?),
+        )
+        .await?;
+    let error: JSONRPCError = timeout(
+        READ_TIMEOUT,
+        app.read_stream_until_error_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    assert_eq!(
+        error.error.message,
+        "queue cannot contain more than 100 submissions"
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn idle_queue_dispatch_preserves_client_id() -> Result<()> {
+    let responses = vec![create_final_assistant_message_sse_response("queued done")?];
+    let (mut app, _codex_home, server) = queue_app(responses).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let queued_submission = ThreadQueueAddParams {
+        thread_id: thread_id.clone(),
+        input: vec![text("durable queued message")],
+        client_user_message_id: "stable-queued-client-id".to_string(),
+    };
+    let queued = queue_item(&mut app, queued_submission.clone()).await?;
+    assert_eq!(
+        queued.client_user_message_id,
+        queued_submission.client_user_message_id
+    );
+    let started: ItemStartedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("item/started")).await??;
+    let ThreadItem::UserMessage {
+        client_id, content, ..
+    } = started.item
+    else {
+        anyhow::bail!("queued turn did not begin with its user message");
+    };
+    assert_eq!(client_id.as_deref(), Some("stable-queued-client-id"));
+    assert_eq!(content, queued_submission.input);
+    let completed: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+    assert_eq!(completed.thread_id, thread_id);
+    assert_eq!(completed.turn.status, TurnStatus::Completed);
+    assert!(list_queue(&mut app, &thread_id).await?.data.is_empty());
+
+    let requests = server
+        .received_requests()
+        .await
+        .context("mock request capture unavailable")?;
+    let request = requests
+        .iter()
+        .find(|request| request.url.path().ends_with("/responses"))
+        .context("queued turn did not reach the model")?;
+    let body = request.body_json::<Value>()?;
+    assert!(body["input"].to_string().contains("durable queued message"));
+    let metadata_header = request
+        .headers
+        .get("x-codex-turn-metadata")
+        .context("queued model request is missing its x-codex-turn-metadata header")?
+        .to_str()
+        .context("queued turn metadata header is not valid ASCII")?;
+    let metadata: Value = serde_json::from_str(metadata_header)?;
+    assert_eq!(metadata["thread_id"].as_str(), Some(thread_id.as_str()));
+    assert_eq!(
+        metadata["turn_id"].as_str(),
+        Some(completed.turn.id.as_str())
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn cold_thread_resume_dispatches_a_persisted_queued_submission() -> Result<()> {
+    let responses = vec![
+        create_final_assistant_message_sse_response("materialized thread")?,
+        create_final_assistant_message_sse_response("cold-resumed queued message")?,
+    ];
+    let (mut first, codex_home, _server) = queue_app(responses).await?;
+    let thread_id = first
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let _: TurnStartResponse = first
+        .request(|request_id| ClientRequest::TurnStart {
+            request_id,
+            params: TurnStartParams {
+                thread_id: thread_id.clone(),
+                input: vec![text("materialize the thread before restarting")],
+                ..Default::default()
+            },
+        })
+        .await?;
+    let _: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, first.read_notification("turn/completed")).await??;
+    drop(first);
+
+    let mut resumed = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build_initialized()
+        .await?;
+    let queued = queue_item(
+        &mut resumed,
+        ThreadQueueAddParams {
+            client_user_message_id: "cold-resumed-queue-item".to_string(),
+            ..submission(&thread_id, "dispatch this after a cold thread resume")
+        },
+    )
+    .await?;
+    assert_eq!(
+        list_queue(&mut resumed, &thread_id).await?.data,
+        vec![queued]
+    );
+    let request_id = resumed
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread_id.clone(),
+            ..Default::default()
+        })
+        .await?;
+    let response: ThreadResumeResponse =
+        timeout(READ_TIMEOUT, resumed.read_response(request_id)).await??;
+    assert_eq!(thread_id, response.thread.id);
+
+    let started: ItemStartedNotification =
+        timeout(READ_TIMEOUT, resumed.read_notification("item/started")).await??;
+    let ThreadItem::UserMessage {
+        client_id, content, ..
+    } = started.item
+    else {
+        anyhow::bail!("cold resume did not start the persisted queued user message");
+    };
+    assert_eq!(client_id.as_deref(), Some("cold-resumed-queue-item"));
+    assert_eq!(
+        content,
+        vec![text("dispatch this after a cold thread resume")]
+    );
+    let completed: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, resumed.read_notification("turn/completed")).await??;
+    assert_eq!(completed.turn.status, TurnStatus::Completed);
+    assert!(list_queue(&mut resumed, &thread_id).await?.data.is_empty());
+    Ok(())
+}
+
+#[tokio::test]
+async fn interrupt_preserves_queue_and_queue_start_can_resume_a_non_head_item() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let responses = vec![
+        blocked_turn_response()?,
+        create_final_assistant_message_sse_response("first queued message done")?,
+        create_final_assistant_message_sse_response("second queued message done")?,
+        create_final_assistant_message_sse_response("message added after interruption done")?,
+        create_final_assistant_message_sse_response("message added after cold resume done")?,
+    ];
+    let (mut app, codex_home, _server) = queue_app(responses).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?;
+
+    let first = queue_item(&mut app, submission(&thread_id, "first queued message")).await?;
+    let second = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            client_user_message_id: "second-queued-client-id".to_string(),
+            ..submission(&thread_id, "second queued message")
+        },
+    )
+    .await?;
+
+    let _: TurnInterruptResponse = app
+        .request(|request_id| ClientRequest::TurnInterrupt {
+            request_id,
+            params: TurnInterruptParams {
+                thread_id: thread_id.clone(),
+                turn_id: active_turn_id,
+            },
+        })
+        .await?;
+    let interrupted: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+    assert_eq!(interrupted.turn.status, TurnStatus::Interrupted);
+    assert_eq!(
+        vec![first.clone(), second.clone()],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    let added_after_interrupt = queue_item(
+        &mut app,
+        submission(&thread_id, "message added after interruption"),
+    )
+    .await?;
+    assert_eq!(
+        vec![first.clone(), second.clone(), added_after_interrupt.clone(),],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    let metadata_resume_request_id = app
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread_id.clone(),
+            exclude_turns: true,
+            ..Default::default()
+        })
+        .await?;
+    let metadata_resumed: ThreadResumeResponse =
+        timeout(READ_TIMEOUT, app.read_response(metadata_resume_request_id)).await??;
+    assert_eq!(metadata_resumed.thread.id, thread_id);
+    assert_eq!(
+        vec![first.clone(), second.clone(), added_after_interrupt.clone(),],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    let resume_request_id = app
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread_id.clone(),
+            ..Default::default()
+        })
+        .await?;
+    let resumed: ThreadResumeResponse =
+        timeout(READ_TIMEOUT, app.read_response(resume_request_id)).await??;
+    assert_eq!(resumed.thread.id, thread_id);
+    assert_eq!(
+        vec![first.clone(), second.clone(), added_after_interrupt.clone(),],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    drop(app);
+    let mut app = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build_initialized()
+        .await?;
+    let cold_resume_request_id = app
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread_id.clone(),
+            ..Default::default()
+        })
+        .await?;
+    let cold_resumed: ThreadResumeResponse =
+        timeout(READ_TIMEOUT, app.read_response(cold_resume_request_id)).await??;
+    assert_eq!(cold_resumed.thread.id, thread_id);
+
+    let added_after_cold_resume = queue_item(
+        &mut app,
+        submission(&thread_id, "message added after cold resume"),
+    )
+    .await?;
+    assert_eq!(
+        vec![
+            first,
+            second.clone(),
+            added_after_interrupt,
+            added_after_cold_resume
+        ],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    let started: ThreadQueueStartResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueStart {
+            request_id,
+            params: ThreadQueueStartParams {
+                thread_id: thread_id.clone(),
+                queued_submission_id: Some(second.id),
+            },
+        })
+        .await?;
+    for index in 0..4 {
+        let completed: TurnCompletedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+        if index == 0 {
+            assert_eq!(completed.turn.id, started.turn.id);
+        }
+        assert_eq!(completed.turn.status, TurnStatus::Completed);
+    }
+    assert!(list_queue(&mut app, &thread_id).await?.data.is_empty());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_start_while_active_returns_busy_and_preserves_the_queue() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        blocked_turn_response()?,
+        create_final_assistant_message_sse_response("active turn done")?,
+        create_final_assistant_message_sse_response("queued message done")?,
+    ])
+    .await;
+    let (mut app, _codex_home, _server) = queue_app_with_server(server).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let (_, approval_id) = start_blocked_turn(&mut app, &thread_id).await?;
+    let queued = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            client_user_message_id: "active-queued-client-id".to_string(),
+            ..submission(&thread_id, "send this queued message now")
+        },
+    )
+    .await?;
+    assert_eq!(queued.client_user_message_id, "active-queued-client-id");
+
+    let start_request_id = app
+        .send_raw_request("thread/queue/start", Some(json!({ "threadId": thread_id })))
+        .await?;
+    let error: JSONRPCError = timeout(
+        READ_TIMEOUT,
+        app.read_stream_until_error_message(RequestId::Integer(start_request_id)),
+    )
+    .await??;
+    assert_eq!(error.error.code, -32600);
+    assert_eq!(
+        error.error.message,
+        "thread already has an active or pending turn"
+    );
+    assert_eq!(list_queue(&mut app, &thread_id).await?.data, vec![queued]);
+    decline_approval(&mut app, approval_id).await?;
+
+    let started_item = loop {
+        let started_item: ItemStartedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("item/started")).await??;
+        if matches!(
+            &started_item.item,
+            ThreadItem::UserMessage { client_id, .. }
+                if client_id.as_deref() == Some("active-queued-client-id")
+        ) {
+            break started_item;
+        }
+    };
+    let ThreadItem::UserMessage { client_id, .. } = started_item.item else {
+        anyhow::bail!("queued message did not start after the active turn completed");
+    };
+    assert_eq!(client_id.as_deref(), Some("active-queued-client-id"));
+
+    let completed = loop {
+        let completed: TurnCompletedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+        if completed.turn.id == started_item.turn_id {
+            break completed;
+        }
+    };
+    assert_eq!(completed.turn.status, TurnStatus::Completed);
+    assert!(list_queue(&mut app, &thread_id).await?.data.is_empty());
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_start_without_id_starts_the_head_when_idle() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let responses = vec![
+        blocked_turn_response()?,
+        create_final_assistant_message_sse_response("first queued message done")?,
+        create_final_assistant_message_sse_response("second queued message done")?,
+    ];
+    let (mut app, _codex_home, _server) = queue_app(responses).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?;
+    let first = queue_item(&mut app, submission(&thread_id, "first queued message")).await?;
+    let second = queue_item(&mut app, submission(&thread_id, "second queued message")).await?;
+
+    let _: TurnInterruptResponse = app
+        .request(|request_id| ClientRequest::TurnInterrupt {
+            request_id,
+            params: TurnInterruptParams {
+                thread_id: thread_id.clone(),
+                turn_id: active_turn_id,
+            },
+        })
+        .await?;
+    let interrupted: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+    assert_eq!(interrupted.turn.status, TurnStatus::Interrupted);
+    assert_eq!(
+        list_queue(&mut app, &thread_id).await?.data,
+        vec![first.clone(), second]
+    );
+
+    let start_request_id = app
+        .send_raw_request("thread/queue/start", Some(json!({ "threadId": thread_id })))
+        .await?;
+    let started: ThreadQueueStartResponse =
+        timeout(READ_TIMEOUT, app.read_response(start_request_id)).await??;
+    let started_item = loop {
+        let started_item: ItemStartedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("item/started")).await??;
+        if matches!(
+            &started_item.item,
+            ThreadItem::UserMessage { client_id, .. }
+                if client_id.as_deref() == Some(first.client_user_message_id.as_str())
+        ) {
+            break started_item;
+        }
+    };
+    assert_eq!(started_item.turn_id, started.turn.id);
+    for _ in 0..2 {
+        let completed: TurnCompletedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+        assert_eq!(completed.turn.status, TurnStatus::Completed);
+    }
+    assert!(list_queue(&mut app, &thread_id).await?.data.is_empty());
+    Ok(())
+}
+
+#[tokio::test]
+async fn a_new_turn_preserves_queued_messages_until_it_completes() -> Result<()> {
+    skip_if_remote!(
+        Ok(()),
+        "uses a host-local command and cwd fixture unavailable to remote executors"
+    );
+
+    let responses = vec![
+        blocked_turn_response()?,
+        blocked_turn_response()?,
+        create_final_assistant_message_sse_response("new turn done")?,
+        create_final_assistant_message_sse_response("first queued message done")?,
+        create_final_assistant_message_sse_response("second queued message done")?,
+    ];
+    let (mut app, _codex_home, _server) = queue_app(responses).await?;
+    let thread_id = app
+        .start_thread(ThreadStartParams::default())
+        .await?
+        .thread
+        .id;
+    let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?;
+
+    let first = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            client_user_message_id: "first-queued-client-id".to_string(),
+            ..submission(&thread_id, "first queued message")
+        },
+    )
+    .await?;
+    let second = queue_item(
+        &mut app,
+        ThreadQueueAddParams {
+            client_user_message_id: "second-queued-client-id".to_string(),
+            ..submission(&thread_id, "second queued message")
+        },
+    )
+    .await?;
+
+    let _: TurnInterruptResponse = app
+        .request(|request_id| ClientRequest::TurnInterrupt {
+            request_id,
+            params: TurnInterruptParams {
+                thread_id: thread_id.clone(),
+                turn_id: active_turn_id,
+            },
+        })
+        .await?;
+    let interrupted: TurnCompletedNotification =
+        timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+    assert_eq!(interrupted.turn.status, TurnStatus::Interrupted);
+
+    let _: TurnStartResponse = app
+        .request(|request_id| ClientRequest::TurnStart {
+            request_id,
+            params: TurnStartParams {
+                thread_id: thread_id.clone(),
+                input: first.input.clone(),
+                client_user_message_id: Some(first.client_user_message_id.clone()),
+                ..Default::default()
+            },
+        })
+        .await?;
+    let approval = timeout(READ_TIMEOUT, app.read_stream_until_request_message()).await??;
+    let ServerRequest::CommandExecutionRequestApproval {
+        request_id: new_approval_id,
+        ..
+    } = approval
+    else {
+        anyhow::bail!("matching ordinary turn did not request command approval");
+    };
+    assert_eq!(
+        vec![first, second],
+        list_queue(&mut app, &thread_id).await?.data
+    );
+
+    decline_approval(&mut app, new_approval_id).await?;
+    for _ in 0..3 {
+        let completed: TurnCompletedNotification =
+            timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??;
+        assert_eq!(completed.turn.status, TurnStatus::Completed);
+    }
+    assert!(list_queue(&mut app, &thread_id).await?.data.is_empty());
+
+    Ok(())
+}
+
+async fn queue_app(responses: Vec<String>) -> Result<(TestAppServer, TempDir, MockServer)> {
+    let server = create_mock_responses_server_sequence(responses).await;
+    queue_app_with_server(server).await
+}
+
+async fn queue_app_with_server(server: MockServer) -> Result<(TestAppServer, TempDir, MockServer)> {
+    let codex_home = TempDir::new()?;
+    let config = MockResponsesConfig::new(&server.uri())
+        .with_approval_policy("untrusted")
+        .with_root_config(r#"approvals_reviewer = "user""#);
+    config.write(codex_home.path())?;
+    let app = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build_initialized()
+        .await?;
+    Ok((app, codex_home, server))
+}
+
+fn blocked_turn_response() -> Result<String> {
+    #[cfg(target_os = "windows")]
+    let shell_command = vec![
+        "powershell".to_string(),
+        "-Command".to_string(),
+        "Start-Sleep -Seconds 10".to_string(),
+    ];
+    #[cfg(not(target_os = "windows"))]
+    let shell_command = vec![
+        "python3".to_string(),
+        "-c".to_string(),
+        "import time; time.sleep(10)".to_string(),
+    ];
+
+    create_shell_command_sse_response(
+        shell_command,
+        /*workdir*/ None,
+        /*timeout_ms*/ Some(10_000),
+        "queue-blocked-command",
+    )
+}
+
+async fn start_blocked_turn(
+    app: &mut TestAppServer,
+    thread_id: &str,
+) -> Result<(String, RequestId)> {
+    let started: TurnStartResponse = app
+        .request(|request_id| ClientRequest::TurnStart {
+            request_id,
+            params: TurnStartParams {
+                thread_id: thread_id.to_string(),
+                input: vec![text("start an approval-blocked turn")],
+                ..Default::default()
+            },
+        })
+        .await?;
+    let approval = timeout(READ_TIMEOUT, app.read_stream_until_request_message()).await??;
+    let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval else {
+        anyhow::bail!("active turn did not request command approval");
+    };
+    Ok((started.turn.id, request_id))
+}
+
+async fn queue_item(
+    app: &mut TestAppServer,
+    params: ThreadQueueAddParams,
+) -> Result<QueuedSubmission> {
+    let response: ThreadQueueAddResponse = app
+        .request(|request_id| ClientRequest::ThreadQueueAdd { request_id, params })
+        .await?;
+    Ok(response.queued_submission)
+}
+
+async fn list_queue(app: &mut TestAppServer, thread_id: &str) -> Result<ThreadQueueListResponse> {
+    let mut data = Vec::new();
+    let mut cursor = None;
+    loop {
+        let page: ThreadQueueListResponse = app
+            .request(|request_id| ClientRequest::ThreadQueueList {
+                request_id,
+                params: ThreadQueueListParams {
+                    thread_id: thread_id.to_string(),
+                    cursor,
+                    limit: None,
+                },
+            })
+            .await?;
+        data.extend(page.data);
+        cursor = page.next_cursor;
+        if cursor.is_none() {
+            return Ok(ThreadQueueListResponse {
+                data,
+                next_cursor: None,
+            });
+        }
+    }
+}
+
+async fn decline_approval(app: &mut TestAppServer, request_id: RequestId) -> Result<()> {
+    app.send_response(
+        request_id,
+        serde_json::to_value(CommandExecutionRequestApprovalResponse {
+            decision: CommandExecutionApprovalDecision::Decline,
+        })?,
+    )
+    .await
+}
+
+fn submission(thread_id: &str, value: &str) -> ThreadQueueAddParams {
+    ThreadQueueAddParams {
+        thread_id: thread_id.to_string(),
+        input: vec![text(value)],
+        client_user_message_id: format!("queued-{value}"),
+    }
+}
+
+fn text(value: &str) -> UserInput {
+    UserInput::Text {
+        text: value.to_string(),
+        text_elements: Vec::new(),
+    }
+}

codex-rs/core/src/session/mod.rs

@@ -1168,6 +1168,14 @@ impl Session {
         self.services.elicitations.subscribe()
     }
 
+    pub(crate) fn mark_interrupted(&self) {
+        self.agent_status.send_replace(AgentStatus::Interrupted);
+    }
+
+    pub(crate) fn is_interrupted(&self) -> bool {
+        matches!(*self.agent_status.borrow(), AgentStatus::Interrupted)
+    }
+
     pub(crate) fn get_tx_event(&self) -> Sender<Event> {
         self.tx_event.clone()
     }

codex-rs/core/src/tasks/lifecycle.rs

@@ -41,9 +41,18 @@ impl Session {
     }
 
     pub(crate) async fn emit_thread_idle_lifecycle_if_idle(&self, cause: ThreadIdleCause) {
-        if self.active_turn.lock().await.is_some()
-            || self.input_queue.has_trigger_turn_mailbox_items().await
-        {
+        let cause = {
+            let active_turn = self.active_turn.lock().await;
+            if active_turn.is_some() {
+                return;
+            }
+            if self.is_interrupted() {
+                ThreadIdleCause::Interrupted
+            } else {
+                cause
+            }
+        };
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
             return;
         }
 

codex-rs/core/src/tasks/mod.rs

@@ -495,7 +495,7 @@ impl Session {
         let mut aborted_turn = false;
         let mut active_turn_to_clear = None;
         let mut turn_context = None;
-        if let Some(mut active_turn) = self.take_active_turn().await {
+        if let Some(mut active_turn) = self.take_active_turn(&reason).await {
             let task = active_turn.task.take();
             aborted_turn = task.is_some();
             turn_context = task.as_ref().map(|task| Arc::clone(&task.turn_context));
@@ -533,6 +533,12 @@ impl Session {
                 .and_then(|active_turn| active_turn.task.as_ref())
                 .is_some_and(|task| task.turn_context.sub_id == turn_id)
             {
+                if matches!(
+                    reason,
+                    TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited
+                ) {
+                    self.mark_interrupted();
+                }
                 active.take()
             } else {
                 None
@@ -769,7 +775,10 @@ impl Session {
                 turn_id: turn_context.sub_id.clone(),
                 profile,
             });
-        let idle_cause = if matches!(abort_reason.as_ref(), Some(TurnAbortReason::Interrupted)) {
+        let idle_cause = if matches!(
+            abort_reason.as_ref(),
+            Some(TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited)
+        ) {
             ThreadIdleCause::Interrupted
         } else if abort_reason.is_none() && turn_context.terminal_error.lock().await.is_some() {
             ThreadIdleCause::Failed
@@ -836,8 +845,17 @@ impl Session {
         }
     }
 
-    async fn take_active_turn(&self) -> Option<ActiveTurn> {
+    async fn take_active_turn(&self, reason: &TurnAbortReason) -> Option<ActiveTurn> {
         let mut active = self.active_turn.lock().await;
+        if matches!(
+            reason,
+            TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited
+        ) && active
+            .as_ref()
+            .is_some_and(|active_turn| active_turn.task.is_some())
+        {
+            self.mark_interrupted();
+        }
         active.take()
     }
 

codex-rs/ext/queue/src/service.rs

@@ -4,7 +4,6 @@ use std::sync::Mutex as StdMutex;
 use std::sync::Weak;
 
 use codex_core::CodexThread;
-use codex_core::NotSubmittedReason;
 use codex_core::StartIfIdleSubmission;
 use codex_core::ThreadManager;
 use codex_core::TurnInput;
@@ -44,14 +43,12 @@ pub struct QueuedItem {
 pub enum QueueServiceError {
     #[error("queue storage failed: {0}")]
     Storage(#[from] ThreadStoreError),
-    #[error("queued item payload is invalid: {0}")]
+    #[error("queued submission payload is invalid: {0}")]
     InvalidPayload(#[from] serde_json::Error),
     #[error("local queued attachment is invalid: {0}")]
     InvalidAttachment(#[from] std::io::Error),
     #[error("Core failed to submit queued user message: {0}")]
     CoreSubmissionError(#[from] CodexErr),
-    #[error("Core declined to start queued user message: {0:?}")]
-    NotStarted(NotSubmittedReason),
     #[error("only user input can be added to the user-message queue")]
     InvalidInput,
     #[error(
@@ -107,15 +104,29 @@ impl QueuedItemService {
     ) -> Result<QueuedItem, QueueServiceError> {
         let input = prepare_queued_user_input(input).await?;
         let payload = serde_json::to_string(&input)?;
-        let item = queued_item_from_record(self.queue.enqueue(thread_id, payload).await?)?;
-        self.emit_changed(thread_id);
+        let item = {
+            let _dispatch_guard = self.dispatch_guard(thread_id).await;
+            let item = queued_item_from_record(self.queue.enqueue(thread_id, payload).await?)?;
+            self.emit_changed(thread_id);
+            item
+        };
         self.wake_if_loaded(thread_id).await;
         Ok(item)
     }
 
     pub async fn list(&self, thread_id: ThreadId) -> Result<Vec<QueuedItem>, QueueServiceError> {
+        self.list_page(thread_id, /*offset*/ 0, MAX_QUEUE_ITEMS)
+            .await
+    }
+
+    pub async fn list_page(
+        &self,
+        thread_id: ThreadId,
+        offset: usize,
+        limit: usize,
+    ) -> Result<Vec<QueuedItem>, QueueServiceError> {
         self.queue
-            .list_page(thread_id, /*offset*/ 0, MAX_QUEUE_ITEMS)
+            .list_page(thread_id, offset, limit)
             .await?
             .into_iter()
             .map(queued_item_from_record)
@@ -129,9 +140,7 @@ impl QueuedItemService {
         mut input: TurnInput,
     ) -> Result<Option<QueuedItem>, QueueServiceError> {
         let _dispatch_guard = self.dispatch_guard(thread_id).await;
-        if let TurnInput::UserInput { client_id, .. } = &mut input
-            && client_id.is_none()
-        {
+        if let TurnInput::UserInput { client_id, .. } = &mut input {
             *client_id = self
                 .list(thread_id)
                 .await?
@@ -190,11 +199,11 @@ impl QueuedItemService {
         Ok(())
     }
 
-    /// Starts the exact queued message when idle and removes it after submission.
+    /// Starts the selected queued message only when its thread is idle.
     pub async fn start(
         &self,
         thread: &CodexThread,
-        queued_item_id: String,
+        queued_item_id: Option<String>,
         trace: Option<W3cTraceContext>,
     ) -> Result<StartIfIdleSubmission, QueueServiceError> {
         let thread_id = thread.session_configured().thread_id;
@@ -203,21 +212,23 @@ impl QueuedItemService {
             .list(thread_id)
             .await?
             .into_iter()
-            .find(|item| item.id == queued_item_id)
+            .find(|item| queued_item_id.as_ref().is_none_or(|id| item.id == *id))
             .ok_or_else(|| ThreadStoreError::InvalidRequest {
-                message: format!("queued item not found: {queued_item_id}"),
+                message: queued_item_id.as_ref().map_or_else(
+                    || "queue is empty".to_string(),
+                    |id| format!("queued submission not found: {id}"),
+                ),
             })?;
+        let queued_item_id = item.id.clone();
         let input @ TurnInput::UserInput { .. } = item.input else {
             return Err(QueueServiceError::InvalidInput);
         };
         let submission = thread
             .start_turn_if_idle(TurnInputRequest::new(input).with_trace(trace))
-            .await
-            .map_err(QueueServiceError::CoreSubmissionError)?;
-        if let StartIfIdleSubmission::NotSubmitted { reason } = submission {
-            return Err(QueueServiceError::NotStarted(reason));
+            .await?;
+        if matches!(submission, StartIfIdleSubmission::Started { .. }) {
+            self.delete_locked(thread_id, queued_item_id).await?;
         }
-        self.delete_locked(thread_id, queued_item_id).await?;
         Ok(submission)
     }
 
@@ -307,10 +318,22 @@ impl QueuedItemService {
 }
 
 async fn prepare_queued_user_input(mut input: TurnInput) -> Result<TurnInput, QueueServiceError> {
-    validate_queued_user_input(&input)?;
     let TurnInput::UserInput { content, client_id } = &mut input else {
         return Err(QueueServiceError::InvalidInput);
     };
+    if content.is_empty() {
+        return Err(QueueServiceError::InvalidInput);
+    }
+    let actual_chars: usize = content
+        .iter()
+        .filter_map(|item| match item {
+            UserInput::Text { text, .. } => Some(text.chars().count()),
+            _ => None,
+        })
+        .sum();
+    if actual_chars > MAX_USER_INPUT_TEXT_CHARS {
+        return Err(QueueServiceError::InputTooLarge { actual_chars });
+    }
     client_id.get_or_insert_with(|| Uuid::now_v7().to_string());
     if !content.iter().any(|item| {
         matches!(
@@ -335,33 +358,13 @@ async fn prepare_queued_user_input(mut input: TurnInput) -> Result<TurnInput, Qu
     .map_err(QueueServiceError::InvalidAttachment)
 }
 
-fn validate_queued_user_input(input: &TurnInput) -> Result<(), QueueServiceError> {
-    let TurnInput::UserInput { content, .. } = input else {
-        return Err(QueueServiceError::InvalidInput);
-    };
-    if content.is_empty() {
-        return Err(QueueServiceError::InvalidInput);
-    }
-    let actual_chars: usize = content
-        .iter()
-        .filter_map(|item| match item {
-            UserInput::Text { text, .. } => Some(text.chars().count()),
-            _ => None,
-        })
-        .sum();
-    if actual_chars > MAX_USER_INPUT_TEXT_CHARS {
-        return Err(QueueServiceError::InputTooLarge { actual_chars });
-    }
-    Ok(())
-}
-
 impl<C> ThreadLifecycleContributor<C> for QueuedItemService
 where
     C: Send + Sync + 'static,
 {
     fn on_thread_idle<'a>(&'a self, input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> {
         Box::pin(async move {
-            if input.cause != ThreadIdleCause::Completed {
+            if input.cause == ThreadIdleCause::Interrupted {
                 return;
             }
             let Ok(thread_id) = ThreadId::from_string(input.thread_store.level_id()) else {

codex-rs/ext/queue/tests/queue_service.rs

@@ -226,24 +226,26 @@ async fn queued_input_and_unique_event_ids_round_trip() -> anyhow::Result<()> {
         7,
         uuid::Uuid::parse_str(generated_client_id)?.get_version_num()
     );
-    assert_eq!(vec![first.clone(), second], service.list(thread_id).await?);
+    assert_eq!(
+        vec![first.clone(), second.clone()],
+        service.list(thread_id).await?
+    );
     assert_eq!(input, first.input);
 
-    let events = sink
-        .events
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    assert_eq!(2, events.len());
-    for event in events.iter() {
-        let EventMsg::ThreadQueueChanged(change) = &event.msg else {
-            anyhow::bail!("event is not a queue change");
-        };
-        assert_eq!(
-            serde_json::json!({ "threadId": thread_id }),
-            serde_json::to_value(change)?
-        );
+    {
+        let events = sink
+            .events
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        assert_eq!(2, events.len());
+        for event in events.iter() {
+            let EventMsg::ThreadQueueChanged(change) = &event.msg else {
+                anyhow::bail!("event is not a queue change");
+            };
+            assert_eq!(change.thread_id, thread_id);
+        }
+        assert_ne!(events[0].id, events[1].id);
     }
-    assert_ne!(events[0].id, events[1].id);
     Ok(())
 }
 
@@ -298,10 +300,17 @@ async fn starting_a_selected_item_preserves_the_remaining_queue() -> anyhow::Res
         .await?;
 
     let submission = service
-        .start(test.codex.as_ref(), second.id.clone(), /*trace*/ None)
+        .start(
+            test.codex.as_ref(),
+            Some(second.id.clone()),
+            /*trace*/ None,
+        )
         .await?;
 
-    assert!(matches!(submission, StartIfIdleSubmission::Started { .. }));
+    assert!(matches!(
+        submission,
+        StartIfIdleSubmission::Started { turn_id } if !turn_id.is_empty()
+    ));
     assert_eq!(vec![first], service.list(thread_id).await?);
     wait_for_event_match(test.codex.as_ref(), |event| match event {
         EventMsg::TurnComplete(_) => Some(()),
@@ -358,13 +367,18 @@ async fn starting_a_selected_item_while_active_leaves_it_queued() -> anyhow::Res
     )
     .await?;
 
-    let error = service
-        .start(test.codex.as_ref(), queued.id.clone(), /*trace*/ None)
-        .await
-        .expect_err("active turn should reject explicit queue start");
+    let submission = service
+        .start(
+            test.codex.as_ref(),
+            Some(queued.id.clone()),
+            /*trace*/ None,
+        )
+        .await?;
     assert!(matches!(
-        error,
-        QueueServiceError::NotStarted(NotSubmittedReason::NotIdle)
+        submission,
+        StartIfIdleSubmission::NotSubmitted {
+            reason: NotSubmittedReason::NotIdle
+        }
     ));
     assert_eq!(vec![queued], service.list(thread_id).await?);
 
@@ -379,8 +393,11 @@ async fn starting_a_selected_item_while_active_leaves_it_queued() -> anyhow::Res
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn stopped_turns_leave_queued_messages_in_place() -> anyhow::Result<()> {
+async fn interrupted_turns_pause_queued_messages_but_failed_turns_drain_them() -> anyhow::Result<()>
+{
     let server = start_mock_server().await;
+    let response =
+        responses::mount_sse_once(&server, responses::sse_completed("failed-follow-up")).await;
     let test = test_codex().build_with_auto_env(&server).await?;
     let thread_id = test.session_configured.thread_id;
     let queue = loaded_thread_queue(&test)?;
@@ -390,14 +407,26 @@ async fn stopped_turns_leave_queued_messages_in_place() -> anyhow::Result<()> {
         Arc::new(NoopExtensionEventSink),
     );
     let queued = service
-        .enqueue(thread_id, user_input("keep queued after stopping"))
+        .enqueue(thread_id, user_input("continue after failure"))
         .await?;
 
-    for cause in [ThreadIdleCause::Interrupted, ThreadIdleCause::Failed] {
-        emit_idle_with_cause(&service, thread_id, cause).await;
+    emit_idle_with_cause(&service, thread_id, ThreadIdleCause::Interrupted).await;
+    assert_eq!(vec![queued], service.list(thread_id).await?);
 
-        assert_eq!(vec![queued.clone()], service.list(thread_id).await?);
-    }
+    emit_idle_with_cause(&service, thread_id, ThreadIdleCause::Failed).await;
+    wait_for_event_match(test.codex.as_ref(), |event| {
+        matches!(event, EventMsg::TurnComplete(_)).then_some(())
+    })
+    .await;
+    assert!(service.list(thread_id).await?.is_empty());
+    assert_eq!(
+        Some("continue after failure"),
+        response
+            .single_request()
+            .message_input_texts("user")
+            .last()
+            .map(String::as_str)
+    );
     Ok(())
 }
 
@@ -541,7 +570,7 @@ async fn explicitly_started_rejected_queue_messages_are_consumed() -> anyhow::Re
     let rejected = queue.enqueue(thread_id, user_input("blocked")).await?;
     let submission = tokio::time::timeout(
         Duration::from_secs(10),
-        queue.start(test.codex.as_ref(), rejected.id, /*trace*/ None),
+        queue.start(test.codex.as_ref(), Some(rejected.id), /*trace*/ None),
     )
     .await?
     .expect("explicitly started input should be submitted");

codex-rs/protocol/src/protocol.rs

@@ -58,8 +58,6 @@ use crate::request_user_input::RequestUserInputResponse;
 use crate::turn_input::TurnInputMode;
 use crate::turn_input::TurnInputRequest;
 use crate::turn_input::TurnInputSubmission;
-#[cfg(test)]
-use crate::user_input::UserInput;
 use codex_extension_items::image_generation::ImageGenerationFailure;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_path_uri::PathUri;
@@ -3839,7 +3837,7 @@ pub struct ThreadGoalUpdatedEvent {
     pub goal: ThreadGoal,
 }
 
-#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "protocol/")]
 pub struct ThreadQueueChangedEvent {
@@ -5562,7 +5560,7 @@ mod tests {
 
     #[test]
     fn user_input_text_serializes_empty_text_elements() -> Result<()> {
-        let input = UserInput::Text {
+        let input = crate::user_input::UserInput::Text {
             text: "hello".to_string(),
             text_elements: Vec::new(),
         };

codex-rs/state/src/runtime/queued_items.rs

@@ -122,7 +122,7 @@ impl SqliteQueueStore {
             transaction.rollback().await?;
             return Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidInput,
-                "queue reorder must include every queued item exactly once",
+                "queue reorder must include every queued submission exactly once",
             )
             .into());
         }

codex-rs/thread-store/src/queue_store.rs

@@ -83,7 +83,9 @@ impl QueueStore for LocalQueueStore {
                 .await
                 .map_err(|error| match error.downcast_ref::<sqlx::Error>() {
                     Some(sqlx::Error::RowNotFound) => ThreadStoreError::InvalidRequest {
-                        message: format!("queue cannot contain more than {MAX_QUEUE_ITEMS} items"),
+                        message: format!(
+                            "queue cannot contain more than {MAX_QUEUE_ITEMS} submissions"
+                        ),
                     },
                     _ => ThreadStoreError::Internal {
                         message: format!("queue storage failed: {error}"),

codex-rs/tui/src/app/app_server_event_targets.rs

@@ -68,6 +68,9 @@ pub(super) fn server_notification_thread_target(
         ServerNotification::ThreadGoalCleared(notification) => {
             Some(notification.thread_id.as_str())
         }
+        ServerNotification::ThreadQueueChanged(notification) => {
+            Some(notification.thread_id.as_str())
+        }
         ServerNotification::ThreadSettingsUpdated(notification) => {
             Some(notification.thread_id.as_str())
         }

codex-rs/tui/src/chatwidget/protocol.rs

@@ -199,6 +199,7 @@ impl ChatWidget {
             | ServerNotification::ThreadStarted(_)
             | ServerNotification::ThreadStatusChanged(_)
             | ServerNotification::ThreadReverted(_)
+            | ServerNotification::ThreadQueueChanged(_)
             | ServerNotification::ThreadArchived(_)
             | ServerNotification::ThreadDeleted(_)
             | ServerNotification::ThreadUnarchived(_)