diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index dc5dd6ad60..8ab71b6b9e 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index b0501b0e9e..0116cfc131 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/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": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 92549d74c8..f5d2e32483 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index d266a3e977..8a994843a5 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json new file mode 100644 index 0000000000..a98dc853ec --- /dev/null +++ b/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" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 3ed710efc0..323a512bb5 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -46,6 +46,7 @@ import type { ThreadClosedNotification } from "./v2/ThreadClosedNotification"; import type { ThreadGoalClearedNotification } from "./v2/ThreadGoalClearedNotification"; import type { ThreadGoalUpdatedNotification } from "./v2/ThreadGoalUpdatedNotification"; import type { ThreadNameUpdatedNotification } from "./v2/ThreadNameUpdatedNotification"; +import type { ThreadQueueChangedNotification } from "./v2/ThreadQueueChangedNotification"; import type { ThreadRealtimeClosedNotification } from "./v2/ThreadRealtimeClosedNotification"; import type { ThreadRealtimeErrorNotification } from "./v2/ThreadRealtimeErrorNotification"; import type { ThreadRealtimeItemAddedNotification } from "./v2/ThreadRealtimeItemAddedNotification"; @@ -70,4 +71,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW /** * Notification sent from the server to the client. */ -export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; +export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/QueuedTurn.ts b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedTurn.ts new file mode 100644 index 0000000000..cfbf763f9f --- /dev/null +++ b/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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/QueuedTurnStatus.ts b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedTurnStatus.ts new file mode 100644 index 0000000000..fc62afe712 --- /dev/null +++ b/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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts new file mode 100644 index 0000000000..d407a3abb3 --- /dev/null +++ b/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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnSubmission.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnSubmission.ts new file mode 100644 index 0000000000..99fe427831 --- /dev/null +++ b/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, /** + * Optional JSON Schema used to constrain the final assistant message for + * this turn. + */ +outputSchema: JsonValue | null}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnSubmissionParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnSubmissionParams.ts new file mode 100644 index 0000000000..a36d8ccb94 --- /dev/null +++ b/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, /** + * Optional JSON Schema used to constrain the final assistant message for + * this turn. + */ +outputSchema?: JsonValue | null}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 66b67fb1b4..c7c6e96e33 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/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"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 71c89a185c..70a76e71ac 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/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), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 8157fce69f..2725e7805f 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/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, + #[ts(optional = nullable)] + pub limit: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueListResponse { + pub data: Vec, + pub next_cursor: Option, +} + +#[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, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueReorderResponse { + pub queued_turns: Vec, +} + +#[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, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs index f9bfe21479..8303ed0419 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs +++ b/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, + /// Optional turn-scoped Responses API client metadata. + #[experimental("turn/start.responsesapiClientMetadata")] + pub responsesapi_client_metadata: Option>, + /// Optional client-provided context fragments keyed by an opaque source identifier. + #[experimental("turn/start.additionalContext")] + pub additional_context: Option>, + /// 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>, + /// Optional JSON Schema used to constrain the final assistant message for + /// this turn. + pub output_schema: Option, +} + +#[derive( + Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi, +)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnSubmissionParams { + pub input: Vec, + /// Optional turn-scoped Responses API client metadata. + #[experimental("turn/start.responsesapiClientMetadata")] + #[ts(optional = nullable)] + pub responsesapi_client_metadata: Option>, + /// Optional client-provided context fragments keyed by an opaque source identifier. + #[experimental("turn/start.additionalContext")] + #[ts(optional = nullable)] + pub additional_context: Option>, + /// 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>, + /// Optional JSON Schema used to constrain the final assistant message for + /// this turn. + #[ts(optional = nullable)] + pub output_schema: Option, +} + +impl From 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 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, )] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 2bf6035b27..93e84449c1 100644 --- a/codex-rs/app-server/README.md +++ b/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. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 2dbd17a3d7..430f1a90a1 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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 } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index a110b73191..c28b1d4188 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/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)] diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 5ba6d02ecb..c43c1eaf40 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/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", diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 939c3df519..2d868aeb6d 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/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, pub(super) thread_goal_processor: ThreadGoalRequestProcessor, + pub(super) thread_queue_processor: ThreadQueueRequestProcessor, pub(super) state_db: Option, pub(super) background_tasks: TaskTracker, pub(super) skills_watcher: Arc, @@ -356,6 +357,7 @@ impl ThreadRequestProcessor { thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, thread_goal_processor: ThreadGoalRequestProcessor, + thread_queue_processor: ThreadQueueRequestProcessor, state_db: Option, skills_watcher: Arc, ) -> 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}")); diff --git a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs new file mode 100644 index 0000000000..9692e3fb5f --- /dev/null +++ b/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, + outgoing: Arc, + config_manager: ConfigManager, + state_db: Option, + thread_state_manager: ThreadStateManager, +} + +impl ThreadQueueRequestProcessor { + pub(crate) fn new( + thread_manager: Arc, + outgoing: Arc, + config_manager: ConfigManager, + state_db: Option, + 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, 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, 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::() + .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::, _>>()?; + 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, 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, 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::, _>>()?; + 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) { + 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, 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::from_string(thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}"))) +} + +fn queued_turn_from_state( + record: codex_state::ThreadQueuedTurn, +) -> Result { + 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) -> TurnError { + TurnError { + message: message.into(), + codex_error_info: None, + additional_details: None, + } +} diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 2fbc64c398..70ce7cc322 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/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)); diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 1605983705..d9e7e63edd 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/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; diff --git a/codex-rs/app-server/tests/suite/v2/thread_queue.rs b/codex-rs/app-server/tests/suite/v2/thread_queue.rs new file mode 100644 index 0000000000..463b8569b0 --- /dev/null +++ b/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![first.clone()] + ); + let second_page = + list_queue_page(&mut mcp, &thread.id, first_page.next_cursor, Some(1)).await?; + assert_eq!( + second_page + .data + .into_iter() + .map(|queued_turn| queued_turn.id) + .collect::>(), + vec![second.clone()] + ); + assert_eq!(second_page.next_cursor, None); + + let reorder_request_id = mcp + .send_raw_request( + "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![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 { + 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 { + 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> { + 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, + limit: Option, +) -> Result { + let request_id = mcp + .send_raw_request( + "thread/queue/list", + Some(serde_json::to_value(ThreadQueueListParams { + thread_id: thread_id.to_string(), + cursor, + limit, + })?), + ) + .await?; + 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, +) -> 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 +"# + ), + ) +} diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 2d7fd818c9..4a18d15ac9 100644 --- a/codex-rs/core/config.schema.json +++ b/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" }, diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 116db0a5b5..2f04952dab 100644 --- a/codex-rs/features/src/lib.rs +++ b/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", diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index 5d7087e90b..0973a1f937 100644 --- a/codex-rs/features/src/tests.rs +++ b/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); diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 0b2143b429..22cf432283 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/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()), diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 6d56330367..5215489621 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/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(_)