feat: expose queued turn app-server RPCs

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

@@ -4118,6 +4118,23 @@
       ],
       "type": "object"
     },
+    "TurnSubmissionParams": {
+      "properties": {
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        },
+        "outputSchema": {
+          "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+        }
+      },
+      "required": [
+        "input"
+      ],
+      "type": "object"
+    },
     "UserInput": {
       "oneOf": [
         {

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

@@ -84,6 +84,28 @@
       ],
       "type": "object"
     },
+    "AdditionalContextEntry": {
+      "properties": {
+        "kind": {
+          "$ref": "#/definitions/AdditionalContextKind"
+        },
+        "value": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "kind",
+        "value"
+      ],
+      "type": "object"
+    },
+    "AdditionalContextKind": {
+      "enum": [
+        "untrusted",
+        "application"
+      ],
+      "type": "string"
+    },
     "AdditionalFileSystemPermissions": {
       "properties": {
         "entries": {
@@ -2654,6 +2676,65 @@
         }
       ]
     },
+    "QueuedTurn": {
+      "properties": {
+        "id": {
+          "type": "string"
+        },
+        "status": {
+          "$ref": "#/definitions/QueuedTurnStatus"
+        },
+        "submission": {
+          "$ref": "#/definitions/TurnSubmission"
+        }
+      },
+      "required": [
+        "id",
+        "status",
+        "submission"
+      ],
+      "type": "object"
+    },
+    "QueuedTurnStatus": {
+      "oneOf": [
+        {
+          "properties": {
+            "type": {
+              "enum": [
+                "pending"
+              ],
+              "title": "PendingQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "type"
+          ],
+          "title": "PendingQueuedTurnStatus",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "error": {
+              "$ref": "#/definitions/TurnError"
+            },
+            "type": {
+              "enum": [
+                "failed"
+              ],
+              "title": "FailedQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "error",
+            "type"
+          ],
+          "title": "FailedQueuedTurnStatus",
+          "type": "object"
+        }
+      ]
+    },
     "RateLimitReachedType": {
       "enum": [
         "rate_limit_reached",
@@ -4246,6 +4327,24 @@
       ],
       "type": "object"
     },
+    "ThreadQueueChangedNotification": {
+      "properties": {
+        "queuedTurns": {
+          "items": {
+            "$ref": "#/definitions/QueuedTurn"
+          },
+          "type": "array"
+        },
+        "threadId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "queuedTurns",
+        "threadId"
+      ],
+      "type": "object"
+    },
     "ThreadRealtimeAudioChunk": {
       "description": "EXPERIMENTAL - thread realtime audio chunk.",
       "properties": {
@@ -4819,6 +4918,21 @@
       ],
       "type": "object"
     },
+    "TurnEnvironmentParams": {
+      "properties": {
+        "cwd": {
+          "$ref": "#/definitions/AbsolutePathBuf"
+        },
+        "environmentId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "cwd",
+        "environmentId"
+      ],
+      "type": "object"
+    },
     "TurnError": {
       "properties": {
         "additionalDetails": {
@@ -4947,6 +5061,23 @@
       ],
       "type": "string"
     },
+    "TurnSubmission": {
+      "properties": {
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        },
+        "outputSchema": {
+          "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+        }
+      },
+      "required": [
+        "input"
+      ],
+      "type": "object"
+    },
     "UserInput": {
       "oneOf": [
         {
@@ -5485,6 +5616,26 @@
       "title": "Thread/settings/updatedNotification",
       "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

@@ -4171,6 +4171,26 @@
           "title": "Thread/settings/updatedNotification",
           "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": {
@@ -13268,6 +13288,65 @@
         ],
         "type": "object"
       },
+      "QueuedTurn": {
+        "properties": {
+          "id": {
+            "type": "string"
+          },
+          "status": {
+            "$ref": "#/definitions/v2/QueuedTurnStatus"
+          },
+          "submission": {
+            "$ref": "#/definitions/v2/TurnSubmission"
+          }
+        },
+        "required": [
+          "id",
+          "status",
+          "submission"
+        ],
+        "type": "object"
+      },
+      "QueuedTurnStatus": {
+        "oneOf": [
+          {
+            "properties": {
+              "type": {
+                "enum": [
+                  "pending"
+                ],
+                "title": "PendingQueuedTurnStatusType",
+                "type": "string"
+              }
+            },
+            "required": [
+              "type"
+            ],
+            "title": "PendingQueuedTurnStatus",
+            "type": "object"
+          },
+          {
+            "properties": {
+              "error": {
+                "$ref": "#/definitions/v2/TurnError"
+              },
+              "type": {
+                "enum": [
+                  "failed"
+                ],
+                "title": "FailedQueuedTurnStatusType",
+                "type": "string"
+              }
+            },
+            "required": [
+              "error",
+              "type"
+            ],
+            "title": "FailedQueuedTurnStatus",
+            "type": "object"
+          }
+        ]
+      },
       "RateLimitReachedType": {
         "enum": [
           "rate_limit_reached",
@@ -16982,6 +17061,26 @@
         "title": "ThreadNameUpdatedNotification",
         "type": "object"
       },
+      "ThreadQueueChangedNotification": {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "properties": {
+          "queuedTurns": {
+            "items": {
+              "$ref": "#/definitions/v2/QueuedTurn"
+            },
+            "type": "array"
+          },
+          "threadId": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "queuedTurns",
+          "threadId"
+        ],
+        "title": "ThreadQueueChangedNotification",
+        "type": "object"
+      },
       "ThreadReadParams": {
         "$schema": "http://json-schema.org/draft-07/schema#",
         "properties": {
@@ -18638,6 +18737,40 @@
         "title": "TurnSteerResponse",
         "type": "object"
       },
+      "TurnSubmission": {
+        "properties": {
+          "input": {
+            "items": {
+              "$ref": "#/definitions/v2/UserInput"
+            },
+            "type": "array"
+          },
+          "outputSchema": {
+            "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+          }
+        },
+        "required": [
+          "input"
+        ],
+        "type": "object"
+      },
+      "TurnSubmissionParams": {
+        "properties": {
+          "input": {
+            "items": {
+              "$ref": "#/definitions/v2/UserInput"
+            },
+            "type": "array"
+          },
+          "outputSchema": {
+            "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+          }
+        },
+        "required": [
+          "input"
+        ],
+        "type": "object"
+      },
       "TurnsPage": {
         "properties": {
           "backwardsCursor": {

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

@@ -9797,6 +9797,65 @@
       ],
       "type": "object"
     },
+    "QueuedTurn": {
+      "properties": {
+        "id": {
+          "type": "string"
+        },
+        "status": {
+          "$ref": "#/definitions/QueuedTurnStatus"
+        },
+        "submission": {
+          "$ref": "#/definitions/TurnSubmission"
+        }
+      },
+      "required": [
+        "id",
+        "status",
+        "submission"
+      ],
+      "type": "object"
+    },
+    "QueuedTurnStatus": {
+      "oneOf": [
+        {
+          "properties": {
+            "type": {
+              "enum": [
+                "pending"
+              ],
+              "title": "PendingQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "type"
+          ],
+          "title": "PendingQueuedTurnStatus",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "error": {
+              "$ref": "#/definitions/TurnError"
+            },
+            "type": {
+              "enum": [
+                "failed"
+              ],
+              "title": "FailedQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "error",
+            "type"
+          ],
+          "title": "FailedQueuedTurnStatus",
+          "type": "object"
+        }
+      ]
+    },
     "RateLimitReachedType": {
       "enum": [
         "rate_limit_reached",
@@ -11518,6 +11577,26 @@
           "title": "Thread/settings/updatedNotification",
           "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": {
@@ -14806,6 +14885,26 @@
       "title": "ThreadNameUpdatedNotification",
       "type": "object"
     },
+    "ThreadQueueChangedNotification": {
+      "$schema": "http://json-schema.org/draft-07/schema#",
+      "properties": {
+        "queuedTurns": {
+          "items": {
+            "$ref": "#/definitions/QueuedTurn"
+          },
+          "type": "array"
+        },
+        "threadId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "queuedTurns",
+        "threadId"
+      ],
+      "title": "ThreadQueueChangedNotification",
+      "type": "object"
+    },
     "ThreadReadParams": {
       "$schema": "http://json-schema.org/draft-07/schema#",
       "properties": {
@@ -16462,6 +16561,40 @@
       "title": "TurnSteerResponse",
       "type": "object"
     },
+    "TurnSubmission": {
+      "properties": {
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        },
+        "outputSchema": {
+          "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+        }
+      },
+      "required": [
+        "input"
+      ],
+      "type": "object"
+    },
+    "TurnSubmissionParams": {
+      "properties": {
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        },
+        "outputSchema": {
+          "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+        }
+      },
+      "required": [
+        "input"
+      ],
+      "type": "object"
+    },
     "TurnsPage": {
       "properties": {
         "backwardsCursor": {

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

@@ -0,0 +1,504 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "definitions": {
+    "AbsolutePathBuf": {
+      "description": "A path that is guaranteed to be absolute and normalized (though it is not guaranteed to be canonicalized or exist on the filesystem).\n\nIMPORTANT: When deserializing an `AbsolutePathBuf`, a base path must be set using [AbsolutePathBufGuard::new]. If no base path is set, the deserialization will fail unless the path being deserialized is already absolute.",
+      "type": "string"
+    },
+    "AdditionalContextEntry": {
+      "properties": {
+        "kind": {
+          "$ref": "#/definitions/AdditionalContextKind"
+        },
+        "value": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "kind",
+        "value"
+      ],
+      "type": "object"
+    },
+    "AdditionalContextKind": {
+      "enum": [
+        "untrusted",
+        "application"
+      ],
+      "type": "string"
+    },
+    "ByteRange": {
+      "properties": {
+        "end": {
+          "format": "uint",
+          "minimum": 0.0,
+          "type": "integer"
+        },
+        "start": {
+          "format": "uint",
+          "minimum": 0.0,
+          "type": "integer"
+        }
+      },
+      "required": [
+        "end",
+        "start"
+      ],
+      "type": "object"
+    },
+    "CodexErrorInfo": {
+      "description": "This translation layer make sure that we expose codex error code in camel case.\n\nWhen an upstream HTTP status is available (for example, from the Responses API or a provider), it is forwarded in `httpStatusCode` on the relevant `codexErrorInfo` variant.",
+      "oneOf": [
+        {
+          "enum": [
+            "contextWindowExceeded",
+            "usageLimitExceeded",
+            "serverOverloaded",
+            "cyberPolicy",
+            "internalServerError",
+            "unauthorized",
+            "badRequest",
+            "threadRollbackFailed",
+            "sandboxError",
+            "other"
+          ],
+          "type": "string"
+        },
+        {
+          "additionalProperties": false,
+          "properties": {
+            "httpConnectionFailed": {
+              "properties": {
+                "httpStatusCode": {
+                  "format": "uint16",
+                  "minimum": 0.0,
+                  "type": [
+                    "integer",
+                    "null"
+                  ]
+                }
+              },
+              "type": "object"
+            }
+          },
+          "required": [
+            "httpConnectionFailed"
+          ],
+          "title": "HttpConnectionFailedCodexErrorInfo",
+          "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "Failed to connect to the response SSE stream.",
+          "properties": {
+            "responseStreamConnectionFailed": {
+              "properties": {
+                "httpStatusCode": {
+                  "format": "uint16",
+                  "minimum": 0.0,
+                  "type": [
+                    "integer",
+                    "null"
+                  ]
+                }
+              },
+              "type": "object"
+            }
+          },
+          "required": [
+            "responseStreamConnectionFailed"
+          ],
+          "title": "ResponseStreamConnectionFailedCodexErrorInfo",
+          "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "The response SSE stream disconnected in the middle of a turn before completion.",
+          "properties": {
+            "responseStreamDisconnected": {
+              "properties": {
+                "httpStatusCode": {
+                  "format": "uint16",
+                  "minimum": 0.0,
+                  "type": [
+                    "integer",
+                    "null"
+                  ]
+                }
+              },
+              "type": "object"
+            }
+          },
+          "required": [
+            "responseStreamDisconnected"
+          ],
+          "title": "ResponseStreamDisconnectedCodexErrorInfo",
+          "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "Reached the retry limit for responses.",
+          "properties": {
+            "responseTooManyFailedAttempts": {
+              "properties": {
+                "httpStatusCode": {
+                  "format": "uint16",
+                  "minimum": 0.0,
+                  "type": [
+                    "integer",
+                    "null"
+                  ]
+                }
+              },
+              "type": "object"
+            }
+          },
+          "required": [
+            "responseTooManyFailedAttempts"
+          ],
+          "title": "ResponseTooManyFailedAttemptsCodexErrorInfo",
+          "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "Returned when `turn/start` or `turn/steer` is submitted while the current active turn cannot accept same-turn steering, for example `/review` or manual `/compact`.",
+          "properties": {
+            "activeTurnNotSteerable": {
+              "properties": {
+                "turnKind": {
+                  "$ref": "#/definitions/NonSteerableTurnKind"
+                }
+              },
+              "required": [
+                "turnKind"
+              ],
+              "type": "object"
+            }
+          },
+          "required": [
+            "activeTurnNotSteerable"
+          ],
+          "title": "ActiveTurnNotSteerableCodexErrorInfo",
+          "type": "object"
+        }
+      ]
+    },
+    "ImageDetail": {
+      "enum": [
+        "auto",
+        "low",
+        "high",
+        "original"
+      ],
+      "type": "string"
+    },
+    "NonSteerableTurnKind": {
+      "enum": [
+        "review",
+        "compact"
+      ],
+      "type": "string"
+    },
+    "QueuedTurn": {
+      "properties": {
+        "id": {
+          "type": "string"
+        },
+        "status": {
+          "$ref": "#/definitions/QueuedTurnStatus"
+        },
+        "submission": {
+          "$ref": "#/definitions/TurnSubmission"
+        }
+      },
+      "required": [
+        "id",
+        "status",
+        "submission"
+      ],
+      "type": "object"
+    },
+    "QueuedTurnStatus": {
+      "oneOf": [
+        {
+          "properties": {
+            "type": {
+              "enum": [
+                "pending"
+              ],
+              "title": "PendingQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "type"
+          ],
+          "title": "PendingQueuedTurnStatus",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "error": {
+              "$ref": "#/definitions/TurnError"
+            },
+            "type": {
+              "enum": [
+                "failed"
+              ],
+              "title": "FailedQueuedTurnStatusType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "error",
+            "type"
+          ],
+          "title": "FailedQueuedTurnStatus",
+          "type": "object"
+        }
+      ]
+    },
+    "TextElement": {
+      "properties": {
+        "byteRange": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/ByteRange"
+            }
+          ],
+          "description": "Byte range in the parent `text` buffer that this element occupies."
+        },
+        "placeholder": {
+          "description": "Optional human-readable placeholder for the element, displayed in the UI.",
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "required": [
+        "byteRange"
+      ],
+      "type": "object"
+    },
+    "TurnEnvironmentParams": {
+      "properties": {
+        "cwd": {
+          "$ref": "#/definitions/AbsolutePathBuf"
+        },
+        "environmentId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "cwd",
+        "environmentId"
+      ],
+      "type": "object"
+    },
+    "TurnError": {
+      "properties": {
+        "additionalDetails": {
+          "default": null,
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "codexErrorInfo": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/CodexErrorInfo"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "message": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "message"
+      ],
+      "type": "object"
+    },
+    "TurnSubmission": {
+      "properties": {
+        "input": {
+          "items": {
+            "$ref": "#/definitions/UserInput"
+          },
+          "type": "array"
+        },
+        "outputSchema": {
+          "description": "Optional JSON Schema used to constrain the final assistant message for this turn."
+        }
+      },
+      "required": [
+        "input"
+      ],
+      "type": "object"
+    },
+    "UserInput": {
+      "oneOf": [
+        {
+          "properties": {
+            "text": {
+              "type": "string"
+            },
+            "text_elements": {
+              "default": [],
+              "description": "UI-defined spans within `text` used to render or persist special elements.",
+              "items": {
+                "$ref": "#/definitions/TextElement"
+              },
+              "type": "array"
+            },
+            "type": {
+              "enum": [
+                "text"
+              ],
+              "title": "TextUserInputType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "text",
+            "type"
+          ],
+          "title": "TextUserInput",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "detail": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ImageDetail"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "default": null
+            },
+            "type": {
+              "enum": [
+                "image"
+              ],
+              "title": "ImageUserInputType",
+              "type": "string"
+            },
+            "url": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "type",
+            "url"
+          ],
+          "title": "ImageUserInput",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "detail": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ImageDetail"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "default": null
+            },
+            "path": {
+              "type": "string"
+            },
+            "type": {
+              "enum": [
+                "localImage"
+              ],
+              "title": "LocalImageUserInputType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "path",
+            "type"
+          ],
+          "title": "LocalImageUserInput",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "path": {
+              "type": "string"
+            },
+            "type": {
+              "enum": [
+                "skill"
+              ],
+              "title": "SkillUserInputType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "name",
+            "path",
+            "type"
+          ],
+          "title": "SkillUserInput",
+          "type": "object"
+        },
+        {
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "path": {
+              "type": "string"
+            },
+            "type": {
+              "enum": [
+                "mention"
+              ],
+              "title": "MentionUserInputType",
+              "type": "string"
+            }
+          },
+          "required": [
+            "name",
+            "path",
+            "type"
+          ],
+          "title": "MentionUserInput",
+          "type": "object"
+        }
+      ]
+    }
+  },
+  "properties": {
+    "queuedTurns": {
+      "items": {
+        "$ref": "#/definitions/QueuedTurn"
+      },
+      "type": "array"
+    },
+    "threadId": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "queuedTurns",
+    "threadId"
+  ],
+  "title": "ThreadQueueChangedNotification",
+  "type": "object"
+}

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

@@ -0,0 +1,7 @@
+// 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 { QueuedTurnStatus } from "./QueuedTurnStatus";
+import type { TurnSubmission } from "./TurnSubmission";
+
+export type QueuedTurn = { id: string, submission: TurnSubmission, status: QueuedTurnStatus, };

codex-rs/app-server-protocol/schema/typescript/v2/QueuedTurnStatus.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 { TurnError } from "./TurnError";
+
+export type QueuedTurnStatus = { "type": "pending" } | { "type": "failed", error: TurnError, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.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 { QueuedTurn } from "./QueuedTurn";
+
+export type ThreadQueueChangedNotification = { threadId: string, queuedTurns: Array<QueuedTurn>, };

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

@@ -0,0 +1,11 @@
+// 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 { JsonValue } from "../serde_json/JsonValue";
+import type { UserInput } from "./UserInput";
+
+export type TurnSubmission = {input: Array<UserInput>, /**
+ * Optional JSON Schema used to constrain the final assistant message for
+ * this turn.
+ */
+outputSchema: JsonValue | null};

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

@@ -0,0 +1,11 @@
+// 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 { JsonValue } from "../serde_json/JsonValue";
+import type { UserInput } from "./UserInput";
+
+export type TurnSubmissionParams = {input: Array<UserInput>, /**
+ * Optional JSON Schema used to constrain the final assistant message for
+ * this turn.
+ */
+outputSchema?: JsonValue | null};

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

@@ -310,6 +310,8 @@ export type { ProcessExitedNotification } from "./ProcessExitedNotification";
 export type { ProcessOutputDeltaNotification } from "./ProcessOutputDeltaNotification";
 export type { ProcessOutputStream } from "./ProcessOutputStream";
 export type { ProcessTerminalSize } from "./ProcessTerminalSize";
+export type { QueuedTurn } from "./QueuedTurn";
+export type { QueuedTurnStatus } from "./QueuedTurnStatus";
 export type { RateLimitReachedType } from "./RateLimitReachedType";
 export type { RateLimitSnapshot } from "./RateLimitSnapshot";
 export type { RateLimitWindow } from "./RateLimitWindow";
@@ -388,6 +390,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";
@@ -453,6 +456,8 @@ export type { TurnStartedNotification } from "./TurnStartedNotification";
 export type { TurnStatus } from "./TurnStatus";
 export type { TurnSteerParams } from "./TurnSteerParams";
 export type { TurnSteerResponse } from "./TurnSteerResponse";
+export type { TurnSubmission } from "./TurnSubmission";
+export type { TurnSubmissionParams } from "./TurnSubmissionParams";
 export type { TurnsPage } from "./TurnsPage";
 export type { UserInput } from "./UserInput";
 export type { WarningNotification } from "./WarningNotification";

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

@@ -509,6 +509,30 @@ 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/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,
+    },
     ThreadMetadataUpdate => "thread/metadata/update" {
         params: v2::ThreadMetadataUpdateParams,
         serialization: thread_id(params.thread_id),
@@ -1485,6 +1509,8 @@ server_notification_definitions! {
     ThreadGoalCleared => "thread/goal/cleared" (v2::ThreadGoalClearedNotification),
     #[experimental("thread/settings/updated")]
     ThreadSettingsUpdated => "thread/settings/updated" (v2::ThreadSettingsUpdatedNotification),
+    #[experimental("thread/queue/changed")]
+    ThreadQueueChanged => "thread/queue/changed" (v2::ThreadQueueChangedNotification),
     ThreadTokenUsageUpdated => "thread/tokenUsage/updated" (v2::ThreadTokenUsageUpdatedNotification),
     TurnStarted => "turn/started" (v2::TurnStartedNotification),
     HookStarted => "hook/started" (v2::HookStartedNotification),

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

@@ -8,7 +8,10 @@ use super::ThreadItem;
 use super::ThreadSource;
 use super::Turn;
 use super::TurnEnvironmentParams;
+use super::TurnError;
 use super::TurnItemsView;
+use super::TurnSubmission;
+use super::TurnSubmissionParams;
 use super::shared::v2_enum_from_core;
 use codex_experimental_api_macros::ExperimentalApi;
 use codex_protocol::config_types::CollaborationMode;
@@ -791,6 +794,95 @@ pub struct ThreadGoalClearResponse {
     pub cleared: bool,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(tag = "type", rename_all = "camelCase")]
+#[ts(tag = "type", export_to = "v2/")]
+pub enum QueuedTurnStatus {
+    Pending,
+    Failed { error: TurnError },
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct QueuedTurn {
+    pub id: String,
+    pub submission: TurnSubmission,
+    pub status: QueuedTurnStatus,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueAddParams {
+    pub thread_id: String,
+    pub submission: TurnSubmissionParams,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueAddResponse {
+    pub queued_turn: QueuedTurn,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueListParams {
+    pub thread_id: String,
+    #[ts(optional = nullable)]
+    pub cursor: Option<String>,
+    #[ts(optional = nullable)]
+    pub limit: Option<u32>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueListResponse {
+    pub data: Vec<QueuedTurn>,
+    pub next_cursor: Option<String>,
+}
+
+#[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_turn_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_turn_ids: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueReorderResponse {
+    pub queued_turns: Vec<QueuedTurn>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadQueueChangedNotification {
+    pub thread_id: String,
+    pub queued_turns: Vec<QueuedTurn>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]

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

@@ -58,6 +58,84 @@ pub struct AdditionalContextEntry {
     pub kind: AdditionalContextKind,
 }
 
+#[derive(
+    Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi,
+)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct TurnSubmission {
+    pub input: Vec<UserInput>,
+    /// Optional turn-scoped Responses API client metadata.
+    #[experimental("turn/start.responsesapiClientMetadata")]
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
+    /// Optional client-provided context fragments keyed by an opaque source identifier.
+    #[experimental("turn/start.additionalContext")]
+    pub additional_context: Option<HashMap<String, AdditionalContextEntry>>,
+    /// Optional turn-scoped environments.
+    ///
+    /// Omitted uses the thread sticky environments. Empty disables
+    /// environment access for this turn. Non-empty selects the first
+    /// environment as the current turn environment for this turn.
+    #[experimental("turn/start.environments")]
+    pub environments: Option<Vec<TurnEnvironmentParams>>,
+    /// Optional JSON Schema used to constrain the final assistant message for
+    /// this turn.
+    pub output_schema: Option<JsonValue>,
+}
+
+#[derive(
+    Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi,
+)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct TurnSubmissionParams {
+    pub input: Vec<UserInput>,
+    /// Optional turn-scoped Responses API client metadata.
+    #[experimental("turn/start.responsesapiClientMetadata")]
+    #[ts(optional = nullable)]
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
+    /// Optional client-provided context fragments keyed by an opaque source identifier.
+    #[experimental("turn/start.additionalContext")]
+    #[ts(optional = nullable)]
+    pub additional_context: Option<HashMap<String, AdditionalContextEntry>>,
+    /// Optional turn-scoped environments.
+    ///
+    /// Omitted uses the thread sticky environments. Empty disables
+    /// environment access for this turn. Non-empty selects the first
+    /// environment as the current turn environment for this turn.
+    #[experimental("turn/start.environments")]
+    #[ts(optional = nullable)]
+    pub environments: Option<Vec<TurnEnvironmentParams>>,
+    /// Optional JSON Schema used to constrain the final assistant message for
+    /// this turn.
+    #[ts(optional = nullable)]
+    pub output_schema: Option<JsonValue>,
+}
+
+impl From<TurnSubmissionParams> for TurnSubmission {
+    fn from(params: TurnSubmissionParams) -> Self {
+        Self {
+            input: params.input,
+            responsesapi_client_metadata: params.responsesapi_client_metadata,
+            additional_context: params.additional_context,
+            environments: params.environments,
+            output_schema: params.output_schema,
+        }
+    }
+}
+
+impl From<TurnSubmission> for TurnSubmissionParams {
+    fn from(submission: TurnSubmission) -> Self {
+        Self {
+            input: submission.input,
+            responsesapi_client_metadata: submission.responsesapi_client_metadata,
+            additional_context: submission.additional_context,
+            environments: submission.environments,
+            output_schema: submission.output_schema,
+        }
+    }
+}
+
 #[derive(
     Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi,
 )]

codex-rs/app-server/README.md

@@ -149,6 +149,11 @@ Example with notification opt-out:
 - `thread/goal/updated` — notification emitted whenever a thread goal changes; includes the full current goal.
 - `thread/goal/cleared` — notification emitted whenever a thread goal is removed.
 - `thread/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/queue/add` — experimental; persist a future turn submission for a loaded thread. The queue stores the turn-scoped submission payload and emits `thread/queue/changed`.
+- `thread/queue/list` — experimental; page through the visible queued turns for a thread with cursor/limit pagination.
+- `thread/queue/delete` — experimental; remove a visible queued turn by id.
+- `thread/queue/reorder` — experimental; replace the visible queue order by queued-turn id.
+- `thread/queue/changed` — experimental notification emitted after visible queue state changes and when a resumed thread replays its queue snapshot.
 - `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.
 - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server keeps the thread loaded and unloads it only after it has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed`.
@@ -582,6 +587,38 @@ Use `thread/goal/clear` to remove the current goal.
 { "method": "thread/goal/cleared", "params": { "threadId": "thr_123" } }
 ```
 
+### Example: Queue a follow-up turn
+
+Experimental clients can store a later submission with `thread/queue/add`. The queue item keeps the message and its turn-scoped context. Dispatch uses the thread's current settings. Clients render queue previews from bounded server snapshots and use paginated `thread/queue/list` when they need the full queue.
+
+```json
+{ "method": "thread/queue/add", "id": 31, "params": {
+    "threadId": "thr_123",
+    "submission": {
+      "input": [{ "type": "text", "text": "Run the follow-up benchmark" }]
+    }
+} }
+{ "id": 31, "result": { "queuedTurn": {
+    "id": "queued_123",
+    "submission": {
+      "input": [{ "type": "text", "text": "Run the follow-up benchmark" }]
+    },
+    "status": { "type": "pending" }
+} } }
+{ "method": "thread/queue/changed", "params": {
+    "threadId": "thr_123",
+    "queuedTurns": [{
+      "id": "queued_123",
+      "submission": {
+        "input": [{ "type": "text", "text": "Run the follow-up benchmark" }]
+      },
+      "status": { "type": "pending" }
+    }]
+} }
+```
+
+Delete or reorder visible queue rows with `thread/queue/delete` and `thread/queue/reorder`; list order is authoritative for rendering.
+
 ### 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/message_processor.rs

@@ -34,6 +34,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;
@@ -180,6 +181,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,
@@ -417,6 +419,27 @@ impl MessageProcessor {
             thread_state_manager.clone(),
             state_db.clone(),
         );
+        let turn_processor = TurnRequestProcessor::new(
+            auth_manager.clone(),
+            Arc::clone(&thread_manager),
+            outgoing.clone(),
+            analytics_events_client.clone(),
+            arg0_paths.clone(),
+            Arc::clone(&config),
+            config_manager.clone(),
+            Arc::clone(&pending_thread_unloads),
+            thread_state_manager.clone(),
+            thread_watch_manager.clone(),
+            Arc::clone(&thread_list_state_permit),
+            Arc::clone(&skills_watcher),
+        );
+        let thread_queue_processor = ThreadQueueRequestProcessor::new(
+            Arc::clone(&thread_manager),
+            outgoing.clone(),
+            config_manager.clone(),
+            state_db.clone(),
+            thread_state_manager.clone(),
+        );
         let thread_processor = ThreadRequestProcessor::new(
             auth_manager.clone(),
             Arc::clone(&thread_manager),
@@ -426,25 +449,12 @@ impl MessageProcessor {
             config_manager.clone(),
             Arc::clone(&thread_store),
             Arc::clone(&pending_thread_unloads),
-            thread_state_manager.clone(),
-            thread_watch_manager.clone(),
-            Arc::clone(&thread_list_state_permit),
-            thread_goal_processor.clone(),
-            state_db,
-            Arc::clone(&skills_watcher),
-        );
-        let turn_processor = TurnRequestProcessor::new(
-            auth_manager.clone(),
-            Arc::clone(&thread_manager),
-            outgoing.clone(),
-            analytics_events_client.clone(),
-            arg0_paths.clone(),
-            Arc::clone(&config),
-            config_manager.clone(),
-            pending_thread_unloads,
             thread_state_manager,
             thread_watch_manager,
-            thread_list_state_permit,
+            Arc::clone(&thread_list_state_permit),
+            thread_goal_processor.clone(),
+            thread_queue_processor.clone(),
+            state_db,
             Arc::clone(&skills_watcher),
         );
         if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) {
@@ -507,6 +517,7 @@ impl MessageProcessor {
             remote_control_processor,
             search_processor,
             thread_goal_processor,
+            thread_queue_processor,
             thread_processor,
             turn_processor,
             windows_sandbox_processor,
@@ -1044,6 +1055,24 @@ impl MessageProcessor {
                     .thread_goal_clear(request_id.clone(), params)
                     .await
             }
+            ClientRequest::ThreadQueueAdd { params, .. } => {
+                self.thread_queue_processor
+                    .thread_queue_add(request_id.clone(), params)
+                    .await
+            }
+            ClientRequest::ThreadQueueList { params, .. } => {
+                self.thread_queue_processor.thread_queue_list(params).await
+            }
+            ClientRequest::ThreadQueueDelete { params, .. } => {
+                self.thread_queue_processor
+                    .thread_queue_delete(request_id.clone(), params)
+                    .await
+            }
+            ClientRequest::ThreadQueueReorder { params, .. } => {
+                self.thread_queue_processor
+                    .thread_queue_reorder(request_id.clone(), params)
+                    .await
+            }
             ClientRequest::ThreadMetadataUpdate { params, .. } => {
                 self.thread_processor.thread_metadata_update(params).await
             }

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

@@ -143,6 +143,8 @@ use codex_app_server_protocol::PluginSource;
 use codex_app_server_protocol::PluginSummary;
 use codex_app_server_protocol::PluginUninstallParams;
 use codex_app_server_protocol::PluginUninstallResponse;
+use codex_app_server_protocol::QueuedTurn;
+use codex_app_server_protocol::QueuedTurnStatus;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ReviewDelivery as ApiReviewDelivery;
 use codex_app_server_protocol::ReviewStartParams;
@@ -203,6 +205,15 @@ use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams;
 use codex_app_server_protocol::ThreadMetadataUpdateParams;
 use codex_app_server_protocol::ThreadMetadataUpdateResponse;
 use codex_app_server_protocol::ThreadNameUpdatedNotification;
+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::ThreadReadParams;
 use codex_app_server_protocol::ThreadReadResponse;
 use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
@@ -255,6 +266,7 @@ use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::TurnSteerParams;
 use codex_app_server_protocol::TurnSteerResponse;
+use codex_app_server_protocol::TurnSubmission;
 use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_app_server_protocol::WindowsSandboxReadiness;
 use codex_app_server_protocol::WindowsSandboxReadinessResponse;
@@ -515,6 +527,7 @@ mod config_errors;
 mod request_errors;
 mod thread_goal_processor;
 mod thread_lifecycle;
+mod thread_queue_processor;
 mod thread_resume_redaction;
 mod thread_summary;
 
@@ -527,6 +540,7 @@ use self::thread_summary::*;
 
 pub(crate) use self::thread_lifecycle::populate_thread_turns_from_history;
 pub(crate) use self::thread_processor::thread_from_stored_thread;
+pub(crate) use self::thread_queue_processor::ThreadQueueRequestProcessor;
 #[cfg(test)]
 pub(crate) use self::thread_summary::read_summary_from_rollout;
 #[cfg(test)]

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

@@ -49,6 +49,7 @@ use serde_json::json;
 use std::path::PathBuf;
 
 const SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT: &[&str] = &[
+    "app_server_queue",
     "apps",
     "memories",
     "mentions_v2",

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

@@ -336,6 +336,7 @@ pub(crate) struct ThreadRequestProcessor {
     pub(super) thread_watch_manager: ThreadWatchManager,
     pub(super) thread_list_state_permit: Arc<Semaphore>,
     pub(super) thread_goal_processor: ThreadGoalRequestProcessor,
+    pub(super) thread_queue_processor: ThreadQueueRequestProcessor,
     pub(super) state_db: Option<StateDbHandle>,
     pub(super) background_tasks: TaskTracker,
     pub(super) skills_watcher: Arc<SkillsWatcher>,
@@ -356,6 +357,7 @@ impl ThreadRequestProcessor {
         thread_watch_manager: ThreadWatchManager,
         thread_list_state_permit: Arc<Semaphore>,
         thread_goal_processor: ThreadGoalRequestProcessor,
+        thread_queue_processor: ThreadQueueRequestProcessor,
         state_db: Option<StateDbHandle>,
         skills_watcher: Arc<SkillsWatcher>,
     ) -> Self {
@@ -372,6 +374,7 @@ impl ThreadRequestProcessor {
             thread_watch_manager,
             thread_list_state_permit,
             thread_goal_processor,
+            thread_queue_processor,
             state_db,
             background_tasks: TaskTracker::new(),
             skills_watcher,
@@ -2709,6 +2712,9 @@ impl ThreadRequestProcessor {
                 self.thread_goal_processor
                     .emit_resume_goal_snapshot_and_continue(thread_id, codex_thread.as_ref())
                     .await;
+                self.thread_queue_processor
+                    .emit_resume_queue_snapshot(thread_id)
+                    .await;
             }
             Err(err) => {
                 let error = internal_error(format!("error resuming thread: {err}"));

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

@@ -0,0 +1,273 @@
+use super::*;
+
+const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25;
+const THREAD_QUEUE_LIST_MAX_LIMIT: usize = 100;
+
+#[derive(Clone)]
+pub(crate) struct ThreadQueueRequestProcessor {
+    thread_manager: Arc<ThreadManager>,
+    outgoing: Arc<OutgoingMessageSender>,
+    config_manager: ConfigManager,
+    state_db: Option<StateDbHandle>,
+    thread_state_manager: ThreadStateManager,
+}
+
+impl ThreadQueueRequestProcessor {
+    pub(crate) fn new(
+        thread_manager: Arc<ThreadManager>,
+        outgoing: Arc<OutgoingMessageSender>,
+        config_manager: ConfigManager,
+        state_db: Option<StateDbHandle>,
+        thread_state_manager: ThreadStateManager,
+    ) -> Self {
+        Self {
+            thread_manager,
+            outgoing,
+            config_manager,
+            state_db,
+            thread_state_manager,
+        }
+    }
+
+    fn state_db(&self) -> Result<&StateDbHandle, JSONRPCErrorError> {
+        self.state_db
+            .as_ref()
+            .ok_or_else(|| internal_error("queued turns require the app-server state db"))
+    }
+
+    async fn require_enabled(&self) -> Result<(), JSONRPCErrorError> {
+        let config = self
+            .config_manager
+            .load_latest_config(/*fallback_cwd*/ None)
+            .await
+            .map_err(|err| internal_error(format!("failed to load app-server config: {err}")))?;
+        if config.features.enabled(Feature::AppServerQueue) {
+            Ok(())
+        } else {
+            Err(invalid_request("app-server queue feature is disabled"))
+        }
+    }
+
+    pub(crate) async fn thread_queue_add(
+        &self,
+        request_id: ConnectionRequestId,
+        params: ThreadQueueAddParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.require_enabled().await?;
+        let thread_id = parse_queue_thread_id(params.thread_id.as_str())?;
+        let thread = self
+            .thread_manager
+            .get_thread(thread_id)
+            .await
+            .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?;
+        if thread.config_snapshot().await.ephemeral {
+            return Err(invalid_request(format!(
+                "ephemeral thread does not support queued turns: {thread_id}"
+            )));
+        }
+        let submission = TurnSubmission::from(params.submission);
+        TurnRequestProcessor::validate_v2_input_limit(&submission.input)?;
+        let payload = serde_json::to_vec(&submission).map_err(|err| {
+            internal_error(format!("failed to serialize queued turn payload: {err}"))
+        })?;
+        let record = self
+            .state_db()?
+            .thread_queue()
+            .append_thread_queued_turn(thread_id, payload.as_slice())
+            .await
+            .map_err(|err| internal_error(format!("failed to add queued turn: {err}")))?;
+        let queued_turn = queued_turn_from_state(record)?;
+        self.outgoing
+            .send_response(
+                request_id,
+                ThreadQueueAddResponse {
+                    queued_turn: queued_turn.clone(),
+                },
+            )
+            .await;
+        self.emit_thread_queue_changed(thread_id).await;
+        Ok(None)
+    }
+
+    pub(crate) async fn thread_queue_list(
+        &self,
+        params: ThreadQueueListParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.require_enabled().await?;
+        let thread_id = parse_queue_thread_id(params.thread_id.as_str())?;
+        let start = match params.cursor {
+            Some(cursor) => cursor
+                .parse::<usize>()
+                .map_err(|_| invalid_request(format!("invalid cursor: {cursor}")))?,
+            None => 0,
+        };
+        let limit = params
+            .limit
+            .unwrap_or(THREAD_QUEUE_LIST_DEFAULT_LIMIT as u32)
+            .clamp(1, THREAD_QUEUE_LIST_MAX_LIMIT as u32) as usize;
+        let records = self
+            .state_db()?
+            .thread_queue()
+            .list_visible_thread_queued_turns_page(thread_id, start, limit.saturating_add(1))
+            .await
+            .map_err(|err| internal_error(format!("failed to read queued turns: {err}")))?;
+        let has_next_page = records.len() > limit;
+        let data = records
+            .into_iter()
+            .take(limit)
+            .map(queued_turn_from_state)
+            .collect::<Result<Vec<_>, _>>()?;
+        let next_cursor = has_next_page.then(|| start.saturating_add(limit).to_string());
+        Ok(Some(ThreadQueueListResponse { data, next_cursor }.into()))
+    }
+
+    pub(crate) async fn thread_queue_delete(
+        &self,
+        request_id: ConnectionRequestId,
+        params: ThreadQueueDeleteParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.require_enabled().await?;
+        let thread_id = parse_queue_thread_id(params.thread_id.as_str())?;
+        let deleted = self
+            .state_db()?
+            .thread_queue()
+            .delete_thread_queued_turn(thread_id, params.queued_turn_id.as_str())
+            .await
+            .map_err(|err| internal_error(format!("failed to delete queued turn: {err}")))?;
+        self.outgoing
+            .send_response(request_id, ThreadQueueDeleteResponse { deleted })
+            .await;
+        if deleted {
+            self.emit_thread_queue_changed(thread_id).await;
+        }
+        Ok(None)
+    }
+
+    pub(crate) async fn thread_queue_reorder(
+        &self,
+        request_id: ConnectionRequestId,
+        params: ThreadQueueReorderParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.require_enabled().await?;
+        let thread_id = parse_queue_thread_id(params.thread_id.as_str())?;
+        let records = self
+            .state_db()?
+            .thread_queue()
+            .reorder_thread_queued_turns(thread_id, params.queued_turn_ids.as_slice())
+            .await
+            .map_err(|err| invalid_request(format!("failed to reorder queued turns: {err}")))?;
+        let queued_turns = records
+            .into_iter()
+            .map(queued_turn_from_state)
+            .collect::<Result<Vec<_>, _>>()?;
+        self.outgoing
+            .send_response(
+                request_id,
+                ThreadQueueReorderResponse {
+                    queued_turns: queued_turns.clone(),
+                },
+            )
+            .await;
+        self.send_thread_queue_changed(thread_id, queued_turns)
+            .await;
+        Ok(None)
+    }
+
+    pub(crate) async fn emit_resume_queue_snapshot(&self, thread_id: ThreadId) {
+        if self.require_enabled().await.is_err() {
+            return;
+        }
+        self.emit_thread_queue_changed(thread_id).await;
+    }
+
+    async fn emit_thread_queue_changed(&self, thread_id: ThreadId) {
+        match self.list_visible_queued_turns(thread_id).await {
+            Ok(queued_turns) => {
+                self.send_thread_queue_changed(thread_id, queued_turns)
+                    .await;
+            }
+            Err(err) => {
+                tracing::warn!("failed to read queue snapshot for thread {thread_id}: {err:?}");
+            }
+        }
+    }
+
+    async fn send_thread_queue_changed(&self, thread_id: ThreadId, queued_turns: Vec<QueuedTurn>) {
+        let subscribed_connection_ids = self
+            .thread_state_manager
+            .subscribed_connection_ids(thread_id)
+            .await;
+        self.outgoing
+            .send_server_notification_to_connections(
+                subscribed_connection_ids.as_slice(),
+                ServerNotification::ThreadQueueChanged(ThreadQueueChangedNotification {
+                    thread_id: thread_id.to_string(),
+                    queued_turns,
+                }),
+            )
+            .await;
+    }
+
+    async fn list_visible_queued_turns(
+        &self,
+        thread_id: ThreadId,
+    ) -> Result<Vec<QueuedTurn>, JSONRPCErrorError> {
+        self.state_db()?
+            .thread_queue()
+            .list_visible_thread_queued_turns_page(
+                thread_id,
+                /*offset*/ 0,
+                THREAD_QUEUE_LIST_MAX_LIMIT,
+            )
+            .await
+            .map_err(|err| internal_error(format!("failed to read queued turns: {err}")))?
+            .into_iter()
+            .map(queued_turn_from_state)
+            .collect()
+    }
+}
+
+fn parse_queue_thread_id(thread_id: &str) -> Result<ThreadId, JSONRPCErrorError> {
+    ThreadId::from_string(thread_id)
+        .map_err(|err| invalid_request(format!("invalid thread id: {err}")))
+}
+
+fn queued_turn_from_state(
+    record: codex_state::ThreadQueuedTurn,
+) -> Result<QueuedTurn, JSONRPCErrorError> {
+    let submission = serde_json::from_slice(record.turn_submission_jsonb.as_slice())
+        .map_err(|err| internal_error(format!("failed to read queued turn payload: {err}")))?;
+    let status = match record.state {
+        codex_state::ThreadQueuedTurnState::Pending => QueuedTurnStatus::Pending,
+        codex_state::ThreadQueuedTurnState::Failed => {
+            let error = record
+                .failure_jsonb
+                .as_deref()
+                .map(serde_json::from_slice)
+                .transpose()
+                .map_err(|err| {
+                    internal_error(format!("failed to read queued turn failure: {err}"))
+                })?
+                .unwrap_or_else(|| turn_error("queued turn dispatch failed"));
+            QueuedTurnStatus::Failed { error }
+        }
+        codex_state::ThreadQueuedTurnState::Dispatching => {
+            return Err(internal_error(
+                "dispatching queued turns are not client-visible",
+            ));
+        }
+    };
+    Ok(QueuedTurn {
+        id: record.queued_turn_id,
+        submission,
+        status,
+    })
+}
+
+fn turn_error(message: impl Into<String>) -> TurnError {
+    TurnError {
+        message: message.into(),
+        codex_error_info: None,
+        additional_details: None,
+    }
+}

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

@@ -369,7 +369,7 @@ impl TurnRequestProcessor {
         error
     }
 
-    fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> {
+    pub(crate) 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/tests/suite/v2/mod.rs

@@ -55,6 +55,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_rollback;

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

@@ -0,0 +1,477 @@
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_final_assistant_message_sse_response;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
+use app_test_support::to_response;
+use codex_app_server_protocol::ClientInfo;
+use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams;
+use codex_app_server_protocol::ExperimentalFeatureEnablementSetResponse;
+use codex_app_server_protocol::InitializeCapabilities;
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::QueuedTurnStatus;
+use codex_app_server_protocol::RequestId;
+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::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnSubmission;
+use codex_app_server_protocol::UserInput as V2UserInput;
+use std::collections::BTreeMap;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
+
+#[tokio::test]
+async fn queue_add_persists_turn_params_and_emits_snapshot() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("unused")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let thread = start_thread(&mut mcp).await?;
+
+    let add_request_id = mcp
+        .send_raw_request(
+            "thread/queue/add",
+            Some(serde_json::to_value(ThreadQueueAddParams {
+                thread_id: thread.id.clone(),
+                submission: text_submission("queued serialized input").into(),
+            })?),
+        )
+        .await?;
+    let add_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(add_request_id)),
+    )
+    .await??;
+    let ThreadQueueAddResponse { queued_turn } = to_response(add_response)?;
+    assert!(matches!(queued_turn.status, QueuedTurnStatus::Pending));
+
+    let notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("thread/queue/changed"),
+    )
+    .await??;
+    let notification: ThreadQueueChangedNotification =
+        serde_json::from_value(notification.params.expect("thread/queue/changed params"))?;
+    assert_eq!(notification.thread_id, thread.id);
+    assert_eq!(notification.queued_turns, vec![queued_turn.clone()]);
+
+    let ThreadQueueListResponse { data, next_cursor } = list_queue_page(
+        &mut mcp, &thread.id, /*cursor*/ None, /*limit*/ None,
+    )
+    .await?;
+    assert_eq!(data, vec![queued_turn]);
+    assert_eq!(next_cursor, None);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_add_rejects_ephemeral_threads() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("unused")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let start_request_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ephemeral: Some(true),
+            ..Default::default()
+        })
+        .await?;
+    let start_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(start_response)?;
+
+    let add_request_id = mcp
+        .send_raw_request(
+            "thread/queue/add",
+            Some(serde_json::to_value(ThreadQueueAddParams {
+                thread_id: thread.id.clone(),
+                submission: text_submission("ephemeral queued turn").into(),
+            })?),
+        )
+        .await?;
+    let add_error: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(add_request_id)),
+    )
+    .await??;
+
+    assert_eq!(add_error.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(
+        add_error.error.message,
+        format!(
+            "ephemeral thread does not support queued turns: {}",
+            thread.id
+        )
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn queue_add_rejects_requests_when_feature_is_disabled() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("unused")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config_without_feature(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let thread = start_thread(&mut mcp).await?;
+
+    let add_request_id = mcp
+        .send_raw_request(
+            "thread/queue/add",
+            Some(serde_json::to_value(ThreadQueueAddParams {
+                thread_id: thread.id,
+                submission: text_submission("disabled queued turn").into(),
+            })?),
+        )
+        .await?;
+    let add_error: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(add_request_id)),
+    )
+    .await??;
+
+    assert_eq!(add_error.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(
+        add_error.error.message,
+        "app-server queue feature is disabled"
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn runtime_feature_enablement_controls_queue_access_without_deleting_rows() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("unused")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config_without_feature(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let thread = start_thread(&mut mcp).await?;
+
+    set_queue_feature(&mut mcp, /*enabled*/ true).await?;
+    let queued_turn_id = queue_turn(&mut mcp, &thread.id, "durable queued turn").await?;
+
+    set_queue_feature(&mut mcp, /*enabled*/ false).await?;
+    let list_request_id = mcp
+        .send_raw_request(
+            "thread/queue/list",
+            Some(serde_json::to_value(ThreadQueueListParams {
+                thread_id: thread.id.clone(),
+                cursor: None,
+                limit: None,
+            })?),
+        )
+        .await?;
+    let list_error: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(list_request_id)),
+    )
+    .await??;
+    assert_eq!(list_error.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(
+        list_error.error.message,
+        "app-server queue feature is disabled"
+    );
+
+    set_queue_feature(&mut mcp, /*enabled*/ true).await?;
+    assert_eq!(
+        list_queue_ids(&mut mcp, &thread.id).await?,
+        vec![queued_turn_id]
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn visible_queue_rows_support_pagination_reorder_and_delete() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        create_final_assistant_message_sse_response("unused")?,
+    ])
+    .await;
+    let codex_home = TempDir::new()?;
+    write_queue_test_config(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    initialize_experimental(&mut mcp).await?;
+    let thread = start_thread(&mut mcp).await?;
+
+    let first = queue_turn(&mut mcp, &thread.id, "first queued").await?;
+    let second = queue_turn(&mut mcp, &thread.id, "second queued").await?;
+    assert_eq!(
+        list_queue_ids(&mut mcp, &thread.id).await?,
+        vec![first.clone(), second.clone()]
+    );
+
+    let first_page = list_queue_page(&mut mcp, &thread.id, /*cursor*/ None, Some(1)).await?;
+    assert_eq!(
+        first_page
+            .data
+            .into_iter()
+            .map(|queued_turn| queued_turn.id)
+            .collect::<Vec<_>>(),
+        vec![first.clone()]
+    );
+    let second_page =
+        list_queue_page(&mut mcp, &thread.id, first_page.next_cursor, Some(1)).await?;
+    assert_eq!(
+        second_page
+            .data
+            .into_iter()
+            .map(|queued_turn| queued_turn.id)
+            .collect::<Vec<_>>(),
+        vec![second.clone()]
+    );
+    assert_eq!(second_page.next_cursor, None);
+
+    let reorder_request_id = mcp
+        .send_raw_request(
+            "thread/queue/reorder",
+            Some(serde_json::to_value(ThreadQueueReorderParams {
+                thread_id: thread.id.clone(),
+                queued_turn_ids: vec![second.clone(), first.clone()],
+            })?),
+        )
+        .await?;
+    let reorder_response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(reorder_request_id)),
+    )
+    .await??;
+    let ThreadQueueReorderResponse { queued_turns } = to_response(reorder_response)?;
+    assert_eq!(
+        queued_turns
+            .into_iter()
+            .map(|queued_turn| queued_turn.id)
+            .collect::<Vec<_>>(),
+        vec![second.clone(), first.clone()]
+    );
+
+    delete_queue_turn(&mut mcp, &thread.id, &second).await?;
+    delete_queue_turn(&mut mcp, &thread.id, &first).await?;
+    assert!(list_queue_ids(&mut mcp, &thread.id).await?.is_empty());
+
+    Ok(())
+}
+
+async fn initialize_experimental(mcp: &mut McpProcess) -> Result<()> {
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_capabilities(
+            ClientInfo {
+                name: "thread-queue-tests".to_string(),
+                title: None,
+                version: "0.0.0".to_string(),
+            },
+            Some(InitializeCapabilities {
+                experimental_api: true,
+                opt_out_notification_methods: None,
+                request_attestation: false,
+            }),
+        ),
+    )
+    .await??;
+    Ok(())
+}
+
+async fn start_thread(mcp: &mut McpProcess) -> Result<codex_app_server_protocol::Thread> {
+    let request_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(response)?;
+    Ok(thread)
+}
+
+async fn queue_turn(mcp: &mut McpProcess, thread_id: &str, text: &str) -> Result<String> {
+    let request_id = mcp
+        .send_raw_request(
+            "thread/queue/add",
+            Some(serde_json::to_value(ThreadQueueAddParams {
+                thread_id: thread_id.to_string(),
+                submission: text_submission(text).into(),
+            })?),
+        )
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let ThreadQueueAddResponse { queued_turn } = to_response(response)?;
+    Ok(queued_turn.id)
+}
+
+async fn list_queue_ids(mcp: &mut McpProcess, thread_id: &str) -> Result<Vec<String>> {
+    let ThreadQueueListResponse { data, .. } =
+        list_queue_page(mcp, thread_id, /*cursor*/ None, /*limit*/ None).await?;
+    Ok(data.into_iter().map(|queued_turn| queued_turn.id).collect())
+}
+
+async fn list_queue_page(
+    mcp: &mut McpProcess,
+    thread_id: &str,
+    cursor: Option<String>,
+    limit: Option<u32>,
+) -> Result<ThreadQueueListResponse> {
+    let request_id = mcp
+        .send_raw_request(
+            "thread/queue/list",
+            Some(serde_json::to_value(ThreadQueueListParams {
+                thread_id: thread_id.to_string(),
+                cursor,
+                limit,
+            })?),
+        )
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    to_response(response)
+}
+
+async fn delete_queue_turn(
+    mcp: &mut McpProcess,
+    thread_id: &str,
+    queued_turn_id: &str,
+) -> Result<()> {
+    let request_id = mcp
+        .send_raw_request(
+            "thread/queue/delete",
+            Some(serde_json::to_value(ThreadQueueDeleteParams {
+                thread_id: thread_id.to_string(),
+                queued_turn_id: queued_turn_id.to_string(),
+            })?),
+        )
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let ThreadQueueDeleteResponse { deleted } = to_response(response)?;
+    assert!(deleted);
+    Ok(())
+}
+
+async fn set_queue_feature(mcp: &mut McpProcess, enabled: bool) -> Result<()> {
+    let request_id = mcp
+        .send_experimental_feature_enablement_set_request(ExperimentalFeatureEnablementSetParams {
+            enablement: BTreeMap::from([("app_server_queue".to_string(), enabled)]),
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let ExperimentalFeatureEnablementSetResponse { enablement } = to_response(response)?;
+    assert_eq!(
+        enablement,
+        BTreeMap::from([("app_server_queue".to_string(), enabled)])
+    );
+    Ok(())
+}
+
+fn text_submission(text: &str) -> TurnSubmission {
+    TurnSubmission {
+        input: vec![V2UserInput::Text {
+            text: text.to_string(),
+            text_elements: Vec::new(),
+        }],
+        ..Default::default()
+    }
+}
+
+fn write_queue_test_config(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> {
+    write_queue_test_config_with_feature(codex_home, server_uri, /*app_server_queue*/ true)
+}
+
+fn write_queue_test_config_with_feature(
+    codex_home: &std::path::Path,
+    server_uri: &str,
+    app_server_queue: bool,
+) -> std::io::Result<()> {
+    write_queue_test_config_with_optional_feature(codex_home, server_uri, Some(app_server_queue))
+}
+
+fn write_queue_test_config_without_feature(
+    codex_home: &std::path::Path,
+    server_uri: &str,
+) -> std::io::Result<()> {
+    write_queue_test_config_with_optional_feature(
+        codex_home, server_uri, /*app_server_queue*/ None,
+    )
+}
+
+fn write_queue_test_config_with_optional_feature(
+    codex_home: &std::path::Path,
+    server_uri: &str,
+    app_server_queue: Option<bool>,
+) -> std::io::Result<()> {
+    let feature_config = app_server_queue
+        .map(|enabled| format!("\n[features]\napp_server_queue = {enabled}\n"))
+        .unwrap_or_default();
+    std::fs::write(
+        codex_home.join("config.toml"),
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "read-only"
+model_provider = "mock_provider"
+{feature_config}
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )
+}

codex-rs/core/config.schema.json

@@ -374,6 +374,9 @@
           "default": null,
           "description": "Optional feature toggles scoped to this profile.",
           "properties": {
+            "app_server_queue": {
+              "type": "boolean"
+            },
             "apply_patch_freeform": {
               "type": "boolean"
             },
@@ -4478,6 +4481,9 @@
       "default": null,
       "description": "Centralized feature flags (new). Prefer this over individual toggles.",
       "properties": {
+        "app_server_queue": {
+          "type": "boolean"
+        },
         "apply_patch_freeform": {
           "type": "boolean"
         },

codex-rs/features/src/lib.rs

@@ -184,6 +184,8 @@ pub enum Feature {
     GuardianApproval,
     /// Enable persisted thread goals and automatic goal continuation.
     Goals,
+    /// Enable durable app-server queued turn submissions.
+    AppServerQueue,
     /// Route MCP tool approval prompts through the MCP elicitation request path.
     ToolCallMcpElicitation,
     /// Prompt Codex Apps connector auth failures through MCP URL elicitations.
@@ -1103,6 +1105,12 @@ pub const FEATURES: &[FeatureSpec] = &[
         stage: Stage::Stable,
         default_enabled: true,
     },
+    FeatureSpec {
+        id: Feature::AppServerQueue,
+        key: "app_server_queue",
+        stage: Stage::UnderDevelopment,
+        default_enabled: false,
+    },
     FeatureSpec {
         id: Feature::CollaborationModes,
         key: "collaboration_modes",

codex-rs/features/src/tests.rs

@@ -136,6 +136,16 @@ fn request_permissions_tool_is_under_development() {
     assert_eq!(Feature::RequestPermissionsTool.default_enabled(), false);
 }
 
+#[test]
+fn app_server_queue_is_under_development() {
+    assert_eq!(Feature::AppServerQueue.stage(), Stage::UnderDevelopment);
+    assert_eq!(Feature::AppServerQueue.default_enabled(), false);
+    assert_eq!(
+        feature_for_key("app_server_queue"),
+        Some(Feature::AppServerQueue)
+    );
+}
+
 #[test]
 fn remote_compaction_v2_is_under_development() {
     assert_eq!(Feature::RemoteCompactionV2.stage(), Stage::UnderDevelopment);

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

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

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

@@ -222,6 +222,7 @@ impl ChatWidget {
             | ServerNotification::ThreadStatusChanged(_)
             | ServerNotification::ThreadArchived(_)
             | ServerNotification::ThreadUnarchived(_)
+            | ServerNotification::ThreadQueueChanged(_)
             | ServerNotification::RawResponseItemCompleted(_)
             | ServerNotification::CommandExecOutputDelta(_)
             | ServerNotification::ProcessOutputDelta(_)