diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 3d786fe1e5..a12721c12f 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -4884,6 +4884,17 @@ ], "type": "object" }, + "ThreadQueueChangedNotification": { + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { @@ -6260,6 +6271,26 @@ "title": "Thread/goal/clearedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/queue/changed" + ], + "title": "Thread/queue/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadQueueChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/queue/changedNotification", + "type": "object" + }, { "properties": { "method": { 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 fe5a258f83..02f6e514c1 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 @@ -4577,6 +4577,26 @@ "title": "Thread/goal/clearedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/queue/changed" + ], + "title": "Thread/queue/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadQueueChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/queue/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -15930,6 +15950,28 @@ ], "type": "object" }, + "QueuedSubmission": { + "properties": { + "clientUserMessageId": { + "type": "string" + }, + "id": { + "type": "string" + }, + "input": { + "items": { + "$ref": "#/definitions/v2/UserInput" + }, + "type": "array" + } + }, + "required": [ + "clientUserMessageId", + "id", + "input" + ], + "type": "object" + }, "RateLimitReachedType": { "enum": [ "rate_limit_reached", @@ -20504,6 +20546,19 @@ "title": "ThreadNameUpdatedNotification", "type": "object" }, + "ThreadQueueChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadQueueChangedNotification", + "type": "object" + }, "ThreadReadParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { 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 82a97d05d9..e5add5a626 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 @@ -12240,6 +12240,28 @@ ], "type": "object" }, + "QueuedSubmission": { + "properties": { + "clientUserMessageId": { + "type": "string" + }, + "id": { + "type": "string" + }, + "input": { + "items": { + "$ref": "#/definitions/UserInput" + }, + "type": "array" + } + }, + "required": [ + "clientUserMessageId", + "id", + "input" + ], + "type": "object" + }, "RateLimitReachedType": { "enum": [ "rate_limit_reached", @@ -14555,6 +14577,26 @@ "title": "Thread/goal/clearedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/queue/changed" + ], + "title": "Thread/queue/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadQueueChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/queue/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -18256,6 +18298,19 @@ "title": "ThreadNameUpdatedNotification", "type": "object" }, + "ThreadQueueChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadQueueChangedNotification", + "type": "object" + }, "ThreadReadParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { 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..25bcdfeb53 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadQueueChangedNotification.json @@ -0,0 +1,13 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadQueueChangedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index cb549a9bd3..ae26f19844 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 4b1eb41a8e..d5e25f6099 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 48fd6d2b0a..d79efc8888 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -51,6 +51,7 @@ import type { ThreadDeletedNotification } from "./v2/ThreadDeletedNotification"; 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"; @@ -77,4 +78,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/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "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/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "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": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "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/progress", "params": ExternalAgentConfigImportProgressNotification } | { "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": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "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/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "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/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "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": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "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/progress", "params": ExternalAgentConfigImportProgressNotification } | { "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": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "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/ServerNotificationEnvelope.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts index 84f048e8ef..130dc46dc7 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts @@ -51,6 +51,7 @@ import type { ThreadDeletedNotification } from "./v2/ThreadDeletedNotification"; 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"; @@ -87,4 +88,4 @@ export type ServerNotificationEnvelope = { * Optional so clients can decode notifications from older app-server * versions. Current app-server versions always populate it. */ -emittedAtMs?: number, } & ({ "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "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/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "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": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "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/progress", "params": ExternalAgentConfigImportProgressNotification } | { "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": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "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 }); +emittedAtMs?: number, } & ({ "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "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/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "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": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "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/progress", "params": ExternalAgentConfigImportProgressNotification } | { "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": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "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/QueuedSubmission.ts b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts new file mode 100644 index 0000000000..f364b8c0de --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { UserInput } from "./UserInput"; + +export type QueuedSubmission = { id: string, input: Array, clientUserMessageId: string, }; 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..55fd1b25e2 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadQueueChangedNotification.ts @@ -0,0 +1,5 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type ThreadQueueChangedNotification = { threadId: string, }; 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 0671837f41..b25c7efa83 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -362,6 +362,7 @@ export type { ProcessExitedNotification } from "./ProcessExitedNotification"; export type { ProcessOutputDeltaNotification } from "./ProcessOutputDeltaNotification"; export type { ProcessOutputStream } from "./ProcessOutputStream"; export type { ProcessTerminalSize } from "./ProcessTerminalSize"; +export type { QueuedSubmission } from "./QueuedSubmission"; export type { RateLimitReachedType } from "./RateLimitReachedType"; export type { RateLimitResetCredit } from "./RateLimitResetCredit"; export type { RateLimitResetCreditStatus } from "./RateLimitResetCreditStatus"; @@ -462,6 +463,7 @@ export type { ThreadMetadataGitInfoUpdateParams } from "./ThreadMetadataGitInfoU export type { ThreadMetadataUpdateParams } from "./ThreadMetadataUpdateParams"; export type { ThreadMetadataUpdateResponse } from "./ThreadMetadataUpdateResponse"; export type { ThreadNameUpdatedNotification } from "./ThreadNameUpdatedNotification"; +export type { ThreadQueueChangedNotification } from "./ThreadQueueChangedNotification"; export type { ThreadReadParams } from "./ThreadReadParams"; export type { ThreadReadResponse } from "./ThreadReadResponse"; export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 813b8b0065..3893b77623 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -574,6 +574,42 @@ client_request_definitions! { serialization: thread_id(params.thread_id), response: v2::ThreadGoalClearResponse, }, + #[experimental("thread/queue/add")] + ThreadQueueAdd => "thread/queue/add" { + params: v2::ThreadQueueAddParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueAddResponse, + }, + #[experimental("thread/queue/list")] + ThreadQueueList => "thread/queue/list" { + params: v2::ThreadQueueListParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueListResponse, + }, + #[experimental("thread/queue/update")] + ThreadQueueUpdate => "thread/queue/update" { + params: v2::ThreadQueueUpdateParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueUpdateResponse, + }, + #[experimental("thread/queue/delete")] + ThreadQueueDelete => "thread/queue/delete" { + params: v2::ThreadQueueDeleteParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueDeleteResponse, + }, + #[experimental("thread/queue/reorder")] + ThreadQueueReorder => "thread/queue/reorder" { + params: v2::ThreadQueueReorderParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueReorderResponse, + }, + #[experimental("thread/queue/start")] + ThreadQueueStart => "thread/queue/start" { + params: v2::ThreadQueueStartParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadQueueStartResponse, + }, ThreadMetadataUpdate => "thread/metadata/update" { params: v2::ThreadMetadataUpdateParams, serialization: thread_id(params.thread_id), @@ -1723,6 +1759,8 @@ server_notification_definitions! { ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification), ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification), ThreadGoalCleared => "thread/goal/cleared" (v2::ThreadGoalClearedNotification), + #[experimental("thread/queue/changed")] + ThreadQueueChanged => "thread/queue/changed" (v2::ThreadQueueChangedNotification), #[experimental("thread/environment/connected")] EnvironmentConnected => "thread/environment/connected" (v2::EnvironmentConnectionNotification), #[experimental("thread/environment/disconnected")] 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 5c8603c5b3..6c409e78f3 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -12,6 +12,7 @@ use super::ThreadSource; use super::Turn; use super::TurnEnvironmentParams; use super::TurnItemsView; +use super::UserInput; use super::shared::v2_enum_from_core; use crate::JsonSchema; use crate::TS; @@ -857,6 +858,113 @@ pub struct ThreadGoalClearResponse { pub cleared: bool, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct QueuedSubmission { + pub id: String, + pub input: Vec, + pub client_user_message_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueAddParams { + pub thread_id: String, + pub input: Vec, + pub client_user_message_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueAddResponse { + pub queued_submission: QueuedSubmission, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueListParams { + pub thread_id: String, + /// Opaque pagination cursor returned by a previous call. + #[ts(optional = nullable)] + pub cursor: Option, + /// Optional page size; defaults to the standard thread-list page size. + #[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, + /// Opaque cursor for the next page, or `null` when no submissions remain. + pub next_cursor: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueUpdateParams { + pub thread_id: String, + pub queued_submission_id: String, + pub input: Vec, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueUpdateResponse { + pub queued_submission: QueuedSubmission, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueDeleteParams { + pub thread_id: String, + pub queued_submission_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueDeleteResponse { + pub deleted: bool, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueReorderParams { + pub thread_id: String, + pub queued_submission_ids: Vec, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueReorderResponse {} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueStartParams { + pub thread_id: String, + #[ts(optional = nullable)] + pub queued_submission_id: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueStartResponse { + pub turn: Turn, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -1768,6 +1876,13 @@ pub struct ThreadGoalClearedNotification { pub thread_id: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadQueueChangedNotification { + pub thread_id: String, +} + /// Deprecated: Use `ContextCompaction` item type instead. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 4f8f97afd7..3fcdd5f05c 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -185,6 +185,13 @@ Example with notification opt-out: - `thread/goal/clear` — clear the current persisted goal for a materialized thread; returns whether a goal was removed and emits `thread/goal/cleared` when state changes. - `thread/goal/updated` — notification emitted whenever a thread goal changes; includes the full current goal. - `thread/goal/cleared` — notification emitted whenever a thread goal is removed. +- `thread/queue/add` — experimental; persist a user turn for automatic FIFO submission when the thread next becomes idle. +- `thread/queue/list` — experimental; return one page of a thread's queued turns. +- `thread/queue/update` — experimental; edit a queued turn while preserving its stable submission ID, client message ID, and position. +- `thread/queue/delete` — experimental; remove a queued turn by submission ID. +- `thread/queue/reorder` — experimental; replace the order of a thread's queued turns. +- `thread/queue/start` — experimental; start the queue head or a selected queued submission when the thread is idle. +- `thread/queue/changed` — experimental notification emitted with the changed `threadId`. - `thread/settings/updated` — experimental notification emitted to subscribed clients when a loaded thread’s effective next-turn settings change; includes `threadId` and the full `threadSettings`. - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`). - `thread/archive` — move a thread’s rollout file into the archived directory and attempt to move any spawned descendant thread rollout files; returns `{}` on success and emits `thread/archived` for each archived thread. @@ -776,6 +783,30 @@ Use `thread/goal/clear` to remove the current goal. { "method": "thread/goal/cleared", "params": { "threadId": "thr_123" } } ``` +### Example: Queue a follow-up user turn (experimental) + +Queued turns require `capabilities.experimentalApi = true`. Use `thread/queue/add` to persist a follow-up while a turn is running. Each thread can queue up to 100 messages, and the server starts the next queued turn when the thread becomes idle. + +A queued submission contains its user input and a required, client-provided `clientUserMessageId`. The server assigns a separate stable submission ID and preserves both IDs when the submission is edited. Application context and Responses API client metadata remain available on ordinary `turn/start`; queued submissions do not persist or replay those optional turn features. + +```json +{ "method": "thread/queue/add", "id": 40, "params": { + "threadId": "thr_123", + "input": [{ "type": "text", "text": "Now fix the failing tests." }], + "clientUserMessageId": "019faba0-0000-7000-8000-000000000003" +} } +{ "id": 40, "result": { "queuedSubmission": { + "id": "019faba0-0000-7000-8000-000000000001", + "input": [{ "type": "text", "text": "Now fix the failing tests." }], + "clientUserMessageId": "019faba0-0000-7000-8000-000000000003" +} } } +{ "method": "thread/queue/changed", "params": { "threadId": "thr_123" } } +``` + +Use `thread/queue/list` to read the ordered queue. Pass optional `cursor` and `limit` values to request a page, and continue with the returned `nextCursor` until it is `null`. Each `thread/queue/changed` notification contains the changed `threadId`; fetch the current pages to refresh the queue. Update a queued turn by passing its `queuedSubmissionId` and replacement `input` to `thread/queue/update`; the submission keeps its IDs and position. Pass that ID to `thread/queue/delete` to remove it, or pass every queued ID in its new order as `queuedSubmissionIds` to `thread/queue/reorder`. + +Completed and failed turns automatically start the next queued submission. Interrupted turns leave the queue paused, including after `thread/resume`. Start the queue head with `thread/queue/start`, or select a queued submission by passing `queuedSubmissionId`. An idle thread starts a new turn and returns it; an active thread returns an invalid-request error and leaves the queue unchanged. The queued submission's client message ID remains stable, and its queue entry is removed when Core accepts the new turn. An ordinary `turn/start` does not consume queued submissions. + ### Example: Archive a thread Use `thread/archive` to move the persisted rollout (stored as a JSONL file on disk) into the archived sessions directory and attempt to move any spawned descendant thread rollouts. diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index 7cf1074b58..3c6cbe2317 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -6,6 +6,7 @@ use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ThreadGoal; use codex_app_server_protocol::ThreadGoalUpdatedNotification; +use codex_app_server_protocol::ThreadQueueChangedNotification; use codex_app_server_protocol::WarningNotification; use codex_core::NewThread; use codex_core::StartThreadOptions; @@ -26,8 +27,8 @@ use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; +use codex_queue_extension::QueuedItemService; use codex_rollout::state_db::StateDbHandle; -use codex_thread_store::QueueStore; use crate::outgoing_message::OutgoingMessageSender; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; @@ -45,8 +46,8 @@ pub(crate) struct ThreadExtensionDependencies { pub(crate) executor_skill_provider: Arc, pub(crate) git_attribution_base_url: String, pub(crate) http_client_factory: HttpClientFactory, - /// Process-scoped persistence backend for queued user messages. - pub(crate) queue_store: Option>, + /// Process-scoped queue shared by idle dispatch and app-server requests. + pub(crate) queue_service: Option>, } pub(crate) fn thread_extensions( @@ -67,15 +68,10 @@ where executor_skill_provider, git_attribution_base_url, http_client_factory, - queue_store, + queue_service, } = dependencies; let mut builder = ExtensionRegistryBuilder::::with_event_sink(Arc::clone(&event_sink)); - if let Some(queue) = queue_store { - let queue_service = Arc::new(codex_queue_extension::QueuedItemService::new( - queue, - thread_manager.clone(), - event_sink, - )); + if let Some(queue_service) = queue_service { codex_queue_extension::install(&mut builder, queue_service); } if let Some(state_db) = state_db { @@ -171,6 +167,40 @@ const EXTENSION_WARNING_SUBSCRIBER_TIMEOUT: Duration = Duration::from_secs(10); impl ExtensionEventSink for AppServerExtensionEventSink { fn emit(&self, event: Event) { match event.msg { + EventMsg::ThreadQueueChanged(queue_event) => { + let thread_id = queue_event.thread_id; + if let Some(listener_command_tx) = self + .thread_state_manager + .current_listener_command_tx(thread_id) + { + let command = ThreadListenerCommand::EmitThreadQueueChanged; + if listener_command_tx.send(command).is_ok() { + return; + } + tracing::warn!( + "failed to enqueue extension queue update for {thread_id}: listener command channel is closed" + ); + } + let outgoing = Arc::clone(&self.outgoing); + let thread_state_manager = self.thread_state_manager.clone(); + tokio::spawn(async move { + let subscribed_connection_ids = thread_state_manager + .subscribed_connection_ids(thread_id) + .await; + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + subscribed_connection_ids, + thread_id, + ); + outgoing + .send_server_notification(ServerNotification::ThreadQueueChanged( + ThreadQueueChangedNotification { + thread_id: thread_id.to_string(), + }, + )) + .await; + }); + } EventMsg::ThreadGoalUpdated(thread_goal_event) => { let thread_id = thread_goal_event.thread_id; let turn_id = thread_goal_event.turn_id; diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 1438e59b48..59c160dd4c 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -108,6 +108,7 @@ fn server_notification_requires_delivery(notification: &ServerNotification) -> b matches!( notification, ServerNotification::TurnCompleted(_) + | ServerNotification::ThreadQueueChanged(_) | ServerNotification::ThreadSettingsUpdated(_) | ServerNotification::ExternalAgentConfigImportCompleted(_) ) @@ -780,6 +781,7 @@ mod tests { use codex_app_server_protocol::ConfigRequirementsReadResponse; use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification; use codex_app_server_protocol::SessionSource as ApiSessionSource; + use codex_app_server_protocol::ThreadQueueChangedNotification; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::Turn; @@ -996,6 +998,11 @@ mod tests { }, }) )); + assert!(server_notification_requires_delivery( + &ServerNotification::ThreadQueueChanged(ThreadQueueChangedNotification { + thread_id: "thread-1".to_string(), + }) + )); assert!(server_notification_requires_delivery( &ServerNotification::ExternalAgentConfigImportCompleted( ExternalAgentConfigImportCompletedNotification { diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index e1bb838f95..148ea6a568 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -339,7 +339,7 @@ enabled = false executor_skill_provider: Arc::clone(&executor_skill_provider), git_attribution_base_url: good_config.chatgpt_base_url.clone(), http_client_factory: good_config.http_client_factory(), - queue_store: None, + queue_service: None, }, ), Arc::new(CodexHomeUserInstructionsProvider::new( diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 0d7480b923..d02be34123 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -37,6 +37,7 @@ use crate::request_processors::ProcessExecRequestProcessor; use crate::request_processors::RemoteControlRequestProcessor; use crate::request_processors::SearchRequestProcessor; use crate::request_processors::ThreadGoalRequestProcessor; +use crate::request_processors::ThreadQueueRequestProcessor; use crate::request_processors::ThreadRequestProcessor; use crate::request_processors::TurnRequestProcessor; use crate::request_processors::WindowsSandboxRequestProcessor; @@ -77,6 +78,7 @@ use codex_protocol::ThreadId; use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::W3cTraceContext; +use codex_queue_extension::QueuedItemService; use codex_rollout::StateDbHandle; use codex_state::log_db::LogDbLayer; use codex_thread_store::LocalQueueStore; @@ -121,6 +123,7 @@ pub(crate) struct MessageProcessor { remote_control_processor: RemoteControlRequestProcessor, search_processor: SearchRequestProcessor, thread_goal_processor: ThreadGoalRequestProcessor, + thread_queue_processor: ThreadQueueRequestProcessor, thread_processor: ThreadRequestProcessor, turn_processor: TurnRequestProcessor, windows_sandbox_processor: WindowsSandboxRequestProcessor, @@ -269,7 +272,17 @@ impl MessageProcessor { ), ); let goal_service = Arc::new(GoalService::new()); + let extension_event_sink = + app_server_extension_event_sink(outgoing.clone(), thread_state_manager.clone()); + let mut queue_service = None; let thread_manager = Arc::new_cyclic(|thread_manager| { + queue_service = queue_store.map(|queue| { + Arc::new(QueuedItemService::new( + queue, + thread_manager.clone(), + Arc::clone(&extension_event_sink), + )) + }); let manager = ThreadManager::new( config.as_ref(), auth_manager.clone(), @@ -280,10 +293,7 @@ impl MessageProcessor { thread_extensions( guardian_agent_spawner(thread_manager.clone()), ThreadExtensionDependencies { - event_sink: app_server_extension_event_sink( - outgoing.clone(), - thread_state_manager.clone(), - ), + event_sink: Arc::clone(&extension_event_sink), auth_manager: auth_manager.clone(), state_db: state_db.clone(), analytics_events_client: analytics_events_client.clone(), @@ -293,7 +303,7 @@ impl MessageProcessor { executor_skill_provider: Arc::clone(&executor_skill_provider), git_attribution_base_url: config.chatgpt_base_url.clone(), http_client_factory: config.http_client_factory(), - queue_store, + queue_service: queue_service.clone(), }, ), Arc::new(CodexHomeUserInstructionsProvider::new( @@ -432,6 +442,12 @@ impl MessageProcessor { state_db.clone(), Arc::clone(&goal_service), ); + let thread_queue_processor = ThreadQueueRequestProcessor::new( + Arc::clone(&thread_manager), + Arc::clone(&thread_store), + outgoing.clone(), + queue_service, + ); let thread_processor = ThreadRequestProcessor::new( auth_manager.clone(), Arc::clone(&thread_manager), @@ -522,6 +538,7 @@ impl MessageProcessor { remote_control_processor, search_processor, thread_goal_processor, + thread_queue_processor, thread_processor, turn_processor, windows_sandbox_processor, @@ -1124,6 +1141,36 @@ impl MessageProcessor { .thread_goal_clear(request_id.clone(), params) .await } + ClientRequest::ThreadQueueAdd { params, .. } => self + .thread_queue_processor + .add(params) + .await + .map(|response| Some(response.into())), + ClientRequest::ThreadQueueList { params, .. } => self + .thread_queue_processor + .list(params) + .await + .map(|response| Some(response.into())), + ClientRequest::ThreadQueueUpdate { params, .. } => self + .thread_queue_processor + .update(params) + .await + .map(|response| Some(response.into())), + ClientRequest::ThreadQueueDelete { params, .. } => self + .thread_queue_processor + .delete(params) + .await + .map(|response| Some(response.into())), + ClientRequest::ThreadQueueReorder { params, .. } => self + .thread_queue_processor + .reorder(params) + .await + .map(|response| Some(response.into())), + ClientRequest::ThreadQueueStart { params, .. } => self + .thread_queue_processor + .start(&request_id, params) + .await + .map(|response| Some(response.into())), ClientRequest::ThreadMetadataUpdate { params, .. } => { self.thread_processor.thread_metadata_update(params).await } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 1f434c193c..616225839e 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -548,6 +548,7 @@ mod search; mod thread_enrichment; mod thread_fork_goal; mod thread_processor; +mod thread_queue_processor; mod thread_sections; mod token_usage_replay; mod turn_processor; @@ -572,6 +573,7 @@ pub(crate) use remote_control_processor::RemoteControlRequestProcessor; pub(crate) use search::SearchRequestProcessor; pub(crate) use thread_goal_processor::ThreadGoalRequestProcessor; pub(crate) use thread_processor::ThreadRequestProcessor; +pub(crate) use thread_queue_processor::ThreadQueueRequestProcessor; pub(crate) use turn_processor::TurnRequestProcessor; pub(crate) use windows_sandbox_processor::WindowsSandboxRequestProcessor; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 34ecd65b48..fa30e16f93 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -1,5 +1,6 @@ use super::*; use crate::extensions::send_thread_warning; +use codex_app_server_protocol::ThreadQueueChangedNotification; use codex_extension_api::ThreadIdleCause; use codex_protocol::config_types::MultiAgentMode; @@ -501,6 +502,23 @@ pub(super) async fn handle_thread_listener_command( )) .await; } + ThreadListenerCommand::EmitThreadQueueChanged => { + let subscribed_connection_ids = thread_state_manager + .subscribed_connection_ids(conversation_id) + .await; + let outgoing = ThreadScopedOutgoingMessageSender::new( + Arc::clone(outgoing), + subscribed_connection_ids, + conversation_id, + ); + outgoing + .send_server_notification(ServerNotification::ThreadQueueChanged( + ThreadQueueChangedNotification { + thread_id: conversation_id.to_string(), + }, + )) + .await; + } ThreadListenerCommand::EmitWarning { message } => { send_thread_warning(outgoing, thread_state_manager, conversation_id, message).await; } 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..1ac74ad878 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs @@ -0,0 +1,336 @@ +use std::sync::Arc; + +use crate::error_code::internal_error; +use crate::error_code::invalid_request; +use crate::outgoing_message::ConnectionRequestId; +use crate::outgoing_message::OutgoingMessageSender; +use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::QueuedSubmission; +use codex_app_server_protocol::ThreadQueueAddParams; +use codex_app_server_protocol::ThreadQueueAddResponse; +use codex_app_server_protocol::ThreadQueueDeleteParams; +use codex_app_server_protocol::ThreadQueueDeleteResponse; +use codex_app_server_protocol::ThreadQueueListParams; +use codex_app_server_protocol::ThreadQueueListResponse; +use codex_app_server_protocol::ThreadQueueReorderParams; +use codex_app_server_protocol::ThreadQueueReorderResponse; +use codex_app_server_protocol::ThreadQueueStartParams; +use codex_app_server_protocol::ThreadQueueStartResponse; +use codex_app_server_protocol::ThreadQueueUpdateParams; +use codex_app_server_protocol::ThreadQueueUpdateResponse; +use codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnItemsView; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; +use codex_core::CodexThread; +use codex_core::NotSubmittedReason; +use codex_core::StartIfIdleSubmission; +use codex_core::ThreadManager; +use codex_core::TurnInput; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_queue_extension::QueueServiceError; +use codex_queue_extension::QueuedItem; +use codex_queue_extension::QueuedItemService; +use codex_thread_store::ReadThreadParams; +use codex_thread_store::ThreadStore; +use codex_thread_store::ThreadStoreError; + +use super::TurnRequestProcessor; +use super::thread_processor::THREAD_LIST_DEFAULT_LIMIT; +use super::thread_processor::THREAD_LIST_MAX_LIMIT; +use super::turn_processor::DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR; +use super::turn_processor::can_accept_direct_input; +use super::turn_processor::validate_user_input_image_urls; + +const DIRECT_INPUT_TO_UNLOADED_SUBAGENT_ERROR: &str = + "direct app-server input is not allowed for unloaded spawned sub-agents"; + +pub(crate) struct ThreadQueueRequestProcessor { + thread_manager: Arc, + thread_store: Arc, + outgoing: Arc, + service: Option>, +} + +impl ThreadQueueRequestProcessor { + pub(crate) fn new( + thread_manager: Arc, + thread_store: Arc, + outgoing: Arc, + service: Option>, + ) -> Self { + Self { + thread_manager, + thread_store, + outgoing, + service, + } + } + + pub(crate) async fn add( + &self, + params: ThreadQueueAddParams, + ) -> Result { + validate_user_input_image_urls(¶ms.input)?; + let (thread_id, loaded_thread, source) = self.require_thread(¶ms.thread_id).await?; + ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?; + let queued_item = self + .service()? + .enqueue( + thread_id, + submission_into_turn_input(params.input, Some(params.client_user_message_id)), + ) + .await + .map_err(queue_error)?; + Ok(ThreadQueueAddResponse { + queued_submission: api_queued_submission(queued_item)?, + }) + } + + pub(crate) async fn list( + &self, + params: ThreadQueueListParams, + ) -> Result { + let (thread_id, _, _) = self.require_thread(¶ms.thread_id).await?; + let offset = params + .cursor + .as_deref() + .map(str::parse::) + .transpose() + .map_err(|error| invalid_request(format!("invalid queue pagination cursor: {error}")))? + .unwrap_or_default(); + let limit = params + .limit + .map(|value| value as usize) + .unwrap_or(THREAD_LIST_DEFAULT_LIMIT) + .clamp(1, THREAD_LIST_MAX_LIMIT); + let mut items = self + .service()? + .list_page(thread_id, offset, limit.saturating_add(1)) + .await + .map_err(queue_error)?; + let next_cursor = if items.len() > limit { + items.truncate(limit); + Some(offset.saturating_add(limit).to_string()) + } else { + None + }; + Ok(ThreadQueueListResponse { + data: items + .into_iter() + .map(api_queued_submission) + .collect::, _>>()?, + next_cursor, + }) + } + + pub(crate) async fn update( + &self, + params: ThreadQueueUpdateParams, + ) -> Result { + validate_user_input_image_urls(¶ms.input)?; + let (thread_id, loaded_thread, source) = self.require_thread(¶ms.thread_id).await?; + ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?; + let queued_item = self + .service()? + .update( + thread_id, + params.queued_submission_id.clone(), + submission_into_turn_input(params.input, /*client_user_message_id*/ None), + ) + .await + .map_err(queue_error)? + .ok_or_else(|| { + invalid_request(format!( + "queued submission not found: {}", + params.queued_submission_id + )) + })?; + Ok(ThreadQueueUpdateResponse { + queued_submission: api_queued_submission(queued_item)?, + }) + } + + pub(crate) async fn delete( + &self, + params: ThreadQueueDeleteParams, + ) -> Result { + let (thread_id, _, _) = self.require_thread(¶ms.thread_id).await?; + let deleted = self + .service()? + .delete(thread_id, params.queued_submission_id) + .await + .map_err(queue_error)?; + Ok(ThreadQueueDeleteResponse { deleted }) + } + + pub(crate) async fn reorder( + &self, + params: ThreadQueueReorderParams, + ) -> Result { + let (thread_id, _, _) = self.require_thread(¶ms.thread_id).await?; + self.service()? + .reorder(thread_id, params.queued_submission_ids) + .await + .map_err(queue_error)?; + Ok(ThreadQueueReorderResponse {}) + } + + pub(crate) async fn start( + &self, + request_id: &ConnectionRequestId, + params: ThreadQueueStartParams, + ) -> Result { + let (_, loaded_thread, source) = self.require_thread(¶ms.thread_id).await?; + ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?; + let thread = loaded_thread + .ok_or_else(|| invalid_request("resume the thread before starting a queued message"))?; + let submission = self + .service()? + .start( + thread.as_ref(), + params.queued_submission_id, + self.outgoing.request_trace_context(request_id).await, + ) + .await + .map_err(queue_error)?; + let turn_id = match submission { + StartIfIdleSubmission::Started { turn_id } => turn_id, + StartIfIdleSubmission::NotSubmitted { + reason: NotSubmittedReason::NotIdle | NotSubmittedReason::PendingTriggerTurn, + } => { + return Err(invalid_request( + "thread already has an active or pending turn", + )); + } + StartIfIdleSubmission::NotSubmitted { reason } => { + return Err(internal_error(format!( + "Core declined to start queued user message: {reason:?}" + ))); + } + }; + self.outgoing + .record_request_turn_id(request_id, &turn_id) + .await; + Ok(ThreadQueueStartResponse { + turn: Turn { + id: turn_id, + items: vec![], + items_view: TurnItemsView::NotLoaded, + error: None, + status: TurnStatus::InProgress, + started_at: None, + completed_at: None, + duration_ms: None, + }, + }) + } + + fn service(&self) -> Result<&QueuedItemService, JSONRPCErrorError> { + self.service + .as_deref() + .ok_or_else(|| invalid_request("user message queue is unavailable")) + } + + async fn require_thread( + &self, + raw_thread_id: &str, + ) -> Result<(ThreadId, Option>, SessionSource), JSONRPCErrorError> { + let thread_id = ThreadId::from_string(raw_thread_id) + .map_err(|error| invalid_request(format!("invalid thread id: {error}")))?; + let (loaded_thread, source) = if let Ok(thread) = + self.thread_manager.get_thread(thread_id).await + { + let snapshot = thread.config_snapshot().await; + if snapshot.ephemeral { + return Err(invalid_request(format!( + "ephemeral thread does not support queued submissions: {thread_id}" + ))); + } + (Some(thread), snapshot.session_source) + } else { + let stored = self + .thread_store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .map_err(|error| match error { + ThreadStoreError::ThreadNotFound { .. } => { + invalid_request(format!("thread not found: {thread_id}")) + } + error => internal_error(format!("failed to read thread: {error}")), + })?; + if stored.archived_at.is_some() { + return Err(invalid_request(format!( + "session {thread_id} is archived. Run `codex unarchive {thread_id}` to unarchive it first." + ))); + } + (None, stored.source) + }; + + Ok((thread_id, loaded_thread, source)) + } +} + +fn ensure_direct_input_allowed( + loaded_thread: Option<&CodexThread>, + source: &SessionSource, +) -> Result<(), JSONRPCErrorError> { + match loaded_thread { + Some(thread) if !can_accept_direct_input(thread.multi_agent_version(), source) => Err( + invalid_request(DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR), + ), + None if matches!( + source, + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) + ) => + { + Err(invalid_request(DIRECT_INPUT_TO_UNLOADED_SUBAGENT_ERROR)) + } + _ => Ok(()), + } +} + +fn submission_into_turn_input( + input: Vec, + client_user_message_id: Option, +) -> TurnInput { + TurnInput::UserInput { + content: input.into_iter().map(UserInput::into_core).collect(), + client_id: client_user_message_id, + } +} + +pub(super) fn queue_error(error: QueueServiceError) -> JSONRPCErrorError { + match error { + QueueServiceError::InputTooLarge { actual_chars } => { + TurnRequestProcessor::input_too_large_error(actual_chars) + } + error @ (QueueServiceError::InvalidInput | QueueServiceError::InvalidAttachment(_)) => { + invalid_request(error.to_string()) + } + QueueServiceError::Storage(ThreadStoreError::InvalidRequest { message }) => { + invalid_request(message) + } + error => internal_error(format!("queued submission operation failed: {error}")), + } +} + +fn api_queued_submission(value: QueuedItem) -> Result { + let TurnInput::UserInput { content, client_id } = value.input else { + return Err(internal_error( + "queued submission does not contain user input", + )); + }; + Ok(QueuedSubmission { + id: value.id, + input: content.into_iter().map(Into::into).collect(), + client_user_message_id: client_id + .ok_or_else(|| internal_error("queued submission is missing its client message id"))?, + }) +} 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 20e9e1bf6e..08de04ffba 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -16,7 +16,7 @@ use codex_skills::system_cache_root_dir; use crate::image_url::REMOTE_IMAGE_URL_ERROR; use crate::image_url::is_remote_image_url; -const DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR: &str = +pub(super) const DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR: &str = "direct app-server input is not allowed for multi-agent v2 sub-agents"; /// Mirrors the direct-input policy in both request validation and thread capability responses. @@ -31,7 +31,9 @@ pub(super) fn can_accept_direct_input( ) } -fn validate_user_input_image_urls(input: &[V2UserInput]) -> Result<(), JSONRPCErrorError> { +pub(super) fn validate_user_input_image_urls( + input: &[V2UserInput], +) -> Result<(), JSONRPCErrorError> { if input.iter().any(|item| { matches!( item, @@ -451,7 +453,7 @@ impl TurnRequestProcessor { .await } - fn input_too_large_error(actual_chars: usize) -> JSONRPCErrorError { + pub(super) fn input_too_large_error(actual_chars: usize) -> JSONRPCErrorError { let mut error = invalid_params(format!( "Input exceeds the maximum length of {MAX_USER_INPUT_TEXT_CHARS} characters." )); @@ -463,7 +465,7 @@ impl TurnRequestProcessor { error } - fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> { + pub(super) fn validate_v2_input_limit(items: &[V2UserInput]) -> Result<(), JSONRPCErrorError> { let actual_chars: usize = items.iter().map(V2UserInput::text_char_count).sum(); if actual_chars > MAX_USER_INPUT_TEXT_CHARS { return Err(Self::input_too_large_error(actual_chars)); diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 1e82d43777..46dc35a36b 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -61,6 +61,8 @@ pub(crate) enum ThreadListenerCommand { turn_id: Option, goal: ThreadGoal, }, + // EmitThreadQueueChanged orders durable queue updates with thread notifications. + EmitThreadQueueChanged, // EmitWarning is used to order extension warnings with other thread notifications. EmitWarning { message: String, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 84149012d2..d473fcfab5 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -90,6 +90,7 @@ mod thread_loaded_list; mod thread_memory_mode_set; mod thread_metadata_update; mod thread_name_websocket; +mod thread_queue; mod thread_read; mod thread_resume; mod thread_revert; 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..97adbf789b --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_queue.rs @@ -0,0 +1,978 @@ +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use app_test_support::MockResponsesConfig; +use app_test_support::TestAppServer; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; +use app_test_support::create_mock_responses_server_sequence_unchecked; +use app_test_support::create_shell_command_sse_response; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::CommandExecutionApprovalDecision; +use codex_app_server_protocol::CommandExecutionRequestApprovalResponse; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::ItemStartedNotification; +use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::QueuedSubmission; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerRequest; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadQueueAddParams; +use codex_app_server_protocol::ThreadQueueAddResponse; +use codex_app_server_protocol::ThreadQueueChangedNotification; +use codex_app_server_protocol::ThreadQueueDeleteParams; +use codex_app_server_protocol::ThreadQueueDeleteResponse; +use codex_app_server_protocol::ThreadQueueListParams; +use codex_app_server_protocol::ThreadQueueListResponse; +use codex_app_server_protocol::ThreadQueueReorderParams; +use codex_app_server_protocol::ThreadQueueReorderResponse; +use codex_app_server_protocol::ThreadQueueStartParams; +use codex_app_server_protocol::ThreadQueueStartResponse; +use codex_app_server_protocol::ThreadQueueUpdateParams; +use codex_app_server_protocol::ThreadQueueUpdateResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; +use core_test_support::skip_if_remote; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use tempfile::TempDir; +use tokio::time::timeout; +use wiremock::MockServer; + +const READ_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 10); + +#[tokio::test] +async fn queue_requires_experimental_handshake() -> Result<()> { + let (mut app, codex_home, _server) = queue_app(Vec::new()).await?; + let thread = app.start_thread(ThreadStartParams::default()).await?.thread; + let queue = list_queue(&mut app, &thread.id).await?; + assert!(queue.data.is_empty()); + drop(app); + + let mut app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build() + .await?; + app.initialize_with_capabilities( + ClientInfo { + name: "queue-experimental-gate".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + Some(InitializeCapabilities::default()), + ) + .await?; + let request_id = app + .send_raw_request("thread/start", Some(json!({}))) + .await?; + let thread = timeout( + READ_TIMEOUT, + app.read_response::(request_id), + ) + .await?? + .thread; + let request_id = app + .send_raw_request( + "thread/queue/list", + Some(serde_json::to_value(ThreadQueueListParams { + thread_id: thread.id, + cursor: None, + limit: None, + })?), + ) + .await?; + let error: JSONRPCError = timeout( + READ_TIMEOUT, + app.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert!(error.error.message.contains("experimental")); + Ok(()) +} + +#[tokio::test] +async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let responses = vec![ + blocked_turn_response()?, + create_final_assistant_message_sse_response("active done")?, + create_final_assistant_message_sse_response("queued done")?, + ]; + let (mut app, _codex_home, _server) = queue_app(responses).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let (_, approval_id) = start_blocked_turn(&mut app, &thread_id).await?; + let first = queue_item( + &mut app, + ThreadQueueAddParams { + client_user_message_id: "first-client-message".to_string(), + ..submission(&thread_id, "first") + }, + ) + .await?; + let second = queue_item(&mut app, submission(&thread_id, "second")).await?; + let first_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + let second_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + first_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + assert_eq!( + second_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + + let updated: ThreadQueueUpdateResponse = app + .request(|request_id| ClientRequest::ThreadQueueUpdate { + request_id, + params: ThreadQueueUpdateParams { + thread_id: thread_id.clone(), + queued_submission_id: first.id.clone(), + input: vec![text("first edited")], + }, + }) + .await?; + assert_eq!(updated.queued_submission.id, first.id); + assert_eq!( + updated.queued_submission.client_user_message_id, + first.client_user_message_id + ); + assert_eq!(updated.queued_submission.input, vec![text("first edited")]); + let update_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + update_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + + let invalid_reorder = app + .send_raw_request( + "thread/queue/reorder", + Some(serde_json::to_value(ThreadQueueReorderParams { + thread_id: thread_id.clone(), + queued_submission_ids: vec![first.id.clone()], + })?), + ) + .await?; + let error: JSONRPCError = timeout( + READ_TIMEOUT, + app.read_stream_until_error_message(RequestId::Integer(invalid_reorder)), + ) + .await??; + assert_eq!( + error.error.message, + "queue reorder must include every queued submission exactly once" + ); + + let _: ThreadQueueReorderResponse = app + .request(|request_id| ClientRequest::ThreadQueueReorder { + request_id, + params: ThreadQueueReorderParams { + thread_id: thread_id.clone(), + queued_submission_ids: vec![second.id.clone(), first.id.clone()], + }, + }) + .await?; + let reorder_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + reorder_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + let deleted: ThreadQueueDeleteResponse = app + .request(|request_id| ClientRequest::ThreadQueueDelete { + request_id, + params: ThreadQueueDeleteParams { + thread_id: thread_id.clone(), + queued_submission_id: second.id, + }, + }) + .await?; + assert!(deleted.deleted); + let delete_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + delete_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + + decline_approval(&mut app, approval_id).await?; + for _ in 0..2 { + let _: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + } + let drain_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!(drain_change, ThreadQueueChangedNotification { thread_id }); + Ok(()) +} + +#[tokio::test] +async fn queue_list_returns_ordered_pages_and_lightweight_notifications() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let (mut app, _codex_home, _server) = queue_app(vec![blocked_turn_response()?]).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let _blocked = start_blocked_turn(&mut app, &thread_id).await?; + let first = queue_item( + &mut app, + submission(&thread_id, "fitting queued submission"), + ) + .await?; + let queued = queue_item( + &mut app, + ThreadQueueAddParams { + thread_id: thread_id.clone(), + input: vec![text(&"x".repeat(64 * 1024))], + client_user_message_id: "oversized-snapshot".to_string(), + }, + ) + .await?; + let initial_change: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + initial_change, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + let changed: ThreadQueueChangedNotification = + timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??; + assert_eq!( + changed, + ThreadQueueChangedNotification { + thread_id: thread_id.clone(), + } + ); + let first_page: ThreadQueueListResponse = app + .request(|request_id| ClientRequest::ThreadQueueList { + request_id, + params: ThreadQueueListParams { + thread_id: thread_id.clone(), + cursor: None, + limit: Some(1), + }, + }) + .await?; + assert_eq!(first_page.data, vec![first.clone()]); + assert_eq!(first_page.next_cursor, Some("1".to_string())); + let second_page: ThreadQueueListResponse = app + .request(|request_id| ClientRequest::ThreadQueueList { + request_id, + params: ThreadQueueListParams { + thread_id: thread_id.clone(), + cursor: first_page.next_cursor, + limit: Some(1), + }, + }) + .await?; + assert_eq!(second_page.data, vec![queued.clone()]); + assert_eq!(second_page.next_cursor, None); + assert_eq!( + list_queue(&mut app, &thread_id).await?.data, + vec![first, queued] + ); + Ok(()) +} + +#[tokio::test] +async fn queue_rejects_messages_after_reaching_its_capacity() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let (mut app, _codex_home, _server) = queue_app(vec![blocked_turn_response()?]).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let _blocked = start_blocked_turn(&mut app, &thread_id).await?; + + for index in 0..100 { + queue_item(&mut app, submission(&thread_id, &format!("queued {index}"))).await?; + } + assert_eq!(list_queue(&mut app, &thread_id).await?.data.len(), 100); + + let request_id = app + .send_raw_request( + "thread/queue/add", + Some(serde_json::to_value(submission( + &thread_id, + "one too many", + ))?), + ) + .await?; + let error: JSONRPCError = timeout( + READ_TIMEOUT, + app.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!( + error.error.message, + "queue cannot contain more than 100 submissions" + ); + Ok(()) +} + +#[tokio::test] +async fn idle_queue_dispatch_preserves_client_id() -> Result<()> { + let responses = vec![create_final_assistant_message_sse_response("queued done")?]; + let (mut app, _codex_home, server) = queue_app(responses).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let queued_submission = ThreadQueueAddParams { + thread_id: thread_id.clone(), + input: vec![text("durable queued message")], + client_user_message_id: "stable-queued-client-id".to_string(), + }; + let queued = queue_item(&mut app, queued_submission.clone()).await?; + assert_eq!( + queued.client_user_message_id, + queued_submission.client_user_message_id + ); + let started: ItemStartedNotification = + timeout(READ_TIMEOUT, app.read_notification("item/started")).await??; + let ThreadItem::UserMessage { + client_id, content, .. + } = started.item + else { + anyhow::bail!("queued turn did not begin with its user message"); + }; + assert_eq!(client_id.as_deref(), Some("stable-queued-client-id")); + assert_eq!(content, queued_submission.input); + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(completed.thread_id, thread_id); + assert_eq!(completed.turn.status, TurnStatus::Completed); + assert!(list_queue(&mut app, &thread_id).await?.data.is_empty()); + + let requests = server + .received_requests() + .await + .context("mock request capture unavailable")?; + let request = requests + .iter() + .find(|request| request.url.path().ends_with("/responses")) + .context("queued turn did not reach the model")?; + let body = request.body_json::()?; + assert!(body["input"].to_string().contains("durable queued message")); + let metadata_header = request + .headers + .get("x-codex-turn-metadata") + .context("queued model request is missing its x-codex-turn-metadata header")? + .to_str() + .context("queued turn metadata header is not valid ASCII")?; + let metadata: Value = serde_json::from_str(metadata_header)?; + assert_eq!(metadata["thread_id"].as_str(), Some(thread_id.as_str())); + assert_eq!( + metadata["turn_id"].as_str(), + Some(completed.turn.id.as_str()) + ); + Ok(()) +} + +#[tokio::test] +async fn cold_thread_resume_dispatches_a_persisted_queued_submission() -> Result<()> { + let responses = vec![ + create_final_assistant_message_sse_response("materialized thread")?, + create_final_assistant_message_sse_response("cold-resumed queued message")?, + ]; + let (mut first, codex_home, _server) = queue_app(responses).await?; + let thread_id = first + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let _: TurnStartResponse = first + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: thread_id.clone(), + input: vec![text("materialize the thread before restarting")], + ..Default::default() + }, + }) + .await?; + let _: TurnCompletedNotification = + timeout(READ_TIMEOUT, first.read_notification("turn/completed")).await??; + drop(first); + + let mut resumed = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized() + .await?; + let queued = queue_item( + &mut resumed, + ThreadQueueAddParams { + client_user_message_id: "cold-resumed-queue-item".to_string(), + ..submission(&thread_id, "dispatch this after a cold thread resume") + }, + ) + .await?; + assert_eq!( + list_queue(&mut resumed, &thread_id).await?.data, + vec![queued] + ); + let request_id = resumed + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }) + .await?; + let response: ThreadResumeResponse = + timeout(READ_TIMEOUT, resumed.read_response(request_id)).await??; + assert_eq!(thread_id, response.thread.id); + + let started: ItemStartedNotification = + timeout(READ_TIMEOUT, resumed.read_notification("item/started")).await??; + let ThreadItem::UserMessage { + client_id, content, .. + } = started.item + else { + anyhow::bail!("cold resume did not start the persisted queued user message"); + }; + assert_eq!(client_id.as_deref(), Some("cold-resumed-queue-item")); + assert_eq!( + content, + vec![text("dispatch this after a cold thread resume")] + ); + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, resumed.read_notification("turn/completed")).await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + assert!(list_queue(&mut resumed, &thread_id).await?.data.is_empty()); + Ok(()) +} + +#[tokio::test] +async fn interrupt_preserves_queue_and_queue_start_can_resume_a_non_head_item() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let responses = vec![ + blocked_turn_response()?, + create_final_assistant_message_sse_response("first queued message done")?, + create_final_assistant_message_sse_response("second queued message done")?, + create_final_assistant_message_sse_response("message added after interruption done")?, + create_final_assistant_message_sse_response("message added after cold resume done")?, + ]; + let (mut app, codex_home, _server) = queue_app(responses).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?; + + let first = queue_item(&mut app, submission(&thread_id, "first queued message")).await?; + let second = queue_item( + &mut app, + ThreadQueueAddParams { + client_user_message_id: "second-queued-client-id".to_string(), + ..submission(&thread_id, "second queued message") + }, + ) + .await?; + + let _: TurnInterruptResponse = app + .request(|request_id| ClientRequest::TurnInterrupt { + request_id, + params: TurnInterruptParams { + thread_id: thread_id.clone(), + turn_id: active_turn_id, + }, + }) + .await?; + let interrupted: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(interrupted.turn.status, TurnStatus::Interrupted); + assert_eq!( + vec![first.clone(), second.clone()], + list_queue(&mut app, &thread_id).await?.data + ); + + let added_after_interrupt = queue_item( + &mut app, + submission(&thread_id, "message added after interruption"), + ) + .await?; + assert_eq!( + vec![first.clone(), second.clone(), added_after_interrupt.clone(),], + list_queue(&mut app, &thread_id).await?.data + ); + + let metadata_resume_request_id = app + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + exclude_turns: true, + ..Default::default() + }) + .await?; + let metadata_resumed: ThreadResumeResponse = + timeout(READ_TIMEOUT, app.read_response(metadata_resume_request_id)).await??; + assert_eq!(metadata_resumed.thread.id, thread_id); + assert_eq!( + vec![first.clone(), second.clone(), added_after_interrupt.clone(),], + list_queue(&mut app, &thread_id).await?.data + ); + + let resume_request_id = app + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }) + .await?; + let resumed: ThreadResumeResponse = + timeout(READ_TIMEOUT, app.read_response(resume_request_id)).await??; + assert_eq!(resumed.thread.id, thread_id); + assert_eq!( + vec![first.clone(), second.clone(), added_after_interrupt.clone(),], + list_queue(&mut app, &thread_id).await?.data + ); + + drop(app); + let mut app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized() + .await?; + let cold_resume_request_id = app + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }) + .await?; + let cold_resumed: ThreadResumeResponse = + timeout(READ_TIMEOUT, app.read_response(cold_resume_request_id)).await??; + assert_eq!(cold_resumed.thread.id, thread_id); + + let added_after_cold_resume = queue_item( + &mut app, + submission(&thread_id, "message added after cold resume"), + ) + .await?; + assert_eq!( + vec![ + first, + second.clone(), + added_after_interrupt, + added_after_cold_resume + ], + list_queue(&mut app, &thread_id).await?.data + ); + + let started: ThreadQueueStartResponse = app + .request(|request_id| ClientRequest::ThreadQueueStart { + request_id, + params: ThreadQueueStartParams { + thread_id: thread_id.clone(), + queued_submission_id: Some(second.id), + }, + }) + .await?; + for index in 0..4 { + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + if index == 0 { + assert_eq!(completed.turn.id, started.turn.id); + } + assert_eq!(completed.turn.status, TurnStatus::Completed); + } + assert!(list_queue(&mut app, &thread_id).await?.data.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn queue_start_while_active_returns_busy_and_preserves_the_queue() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let server = create_mock_responses_server_sequence_unchecked(vec![ + blocked_turn_response()?, + create_final_assistant_message_sse_response("active turn done")?, + create_final_assistant_message_sse_response("queued message done")?, + ]) + .await; + let (mut app, _codex_home, _server) = queue_app_with_server(server).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let (_, approval_id) = start_blocked_turn(&mut app, &thread_id).await?; + let queued = queue_item( + &mut app, + ThreadQueueAddParams { + client_user_message_id: "active-queued-client-id".to_string(), + ..submission(&thread_id, "send this queued message now") + }, + ) + .await?; + assert_eq!(queued.client_user_message_id, "active-queued-client-id"); + + let start_request_id = app + .send_raw_request("thread/queue/start", Some(json!({ "threadId": thread_id }))) + .await?; + let error: JSONRPCError = timeout( + READ_TIMEOUT, + app.read_stream_until_error_message(RequestId::Integer(start_request_id)), + ) + .await??; + assert_eq!(error.error.code, -32600); + assert_eq!( + error.error.message, + "thread already has an active or pending turn" + ); + assert_eq!(list_queue(&mut app, &thread_id).await?.data, vec![queued]); + decline_approval(&mut app, approval_id).await?; + + let started_item = loop { + let started_item: ItemStartedNotification = + timeout(READ_TIMEOUT, app.read_notification("item/started")).await??; + if matches!( + &started_item.item, + ThreadItem::UserMessage { client_id, .. } + if client_id.as_deref() == Some("active-queued-client-id") + ) { + break started_item; + } + }; + let ThreadItem::UserMessage { client_id, .. } = started_item.item else { + anyhow::bail!("queued message did not start after the active turn completed"); + }; + assert_eq!(client_id.as_deref(), Some("active-queued-client-id")); + + let completed = loop { + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + if completed.turn.id == started_item.turn_id { + break completed; + } + }; + assert_eq!(completed.turn.status, TurnStatus::Completed); + assert!(list_queue(&mut app, &thread_id).await?.data.is_empty()); + Ok(()) +} + +#[tokio::test] +async fn queue_start_without_id_starts_the_head_when_idle() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let responses = vec![ + blocked_turn_response()?, + create_final_assistant_message_sse_response("first queued message done")?, + create_final_assistant_message_sse_response("second queued message done")?, + ]; + let (mut app, _codex_home, _server) = queue_app(responses).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?; + let first = queue_item(&mut app, submission(&thread_id, "first queued message")).await?; + let second = queue_item(&mut app, submission(&thread_id, "second queued message")).await?; + + let _: TurnInterruptResponse = app + .request(|request_id| ClientRequest::TurnInterrupt { + request_id, + params: TurnInterruptParams { + thread_id: thread_id.clone(), + turn_id: active_turn_id, + }, + }) + .await?; + let interrupted: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(interrupted.turn.status, TurnStatus::Interrupted); + assert_eq!( + list_queue(&mut app, &thread_id).await?.data, + vec![first.clone(), second] + ); + + let start_request_id = app + .send_raw_request("thread/queue/start", Some(json!({ "threadId": thread_id }))) + .await?; + let started: ThreadQueueStartResponse = + timeout(READ_TIMEOUT, app.read_response(start_request_id)).await??; + let started_item = loop { + let started_item: ItemStartedNotification = + timeout(READ_TIMEOUT, app.read_notification("item/started")).await??; + if matches!( + &started_item.item, + ThreadItem::UserMessage { client_id, .. } + if client_id.as_deref() == Some(first.client_user_message_id.as_str()) + ) { + break started_item; + } + }; + assert_eq!(started_item.turn_id, started.turn.id); + for _ in 0..2 { + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + } + assert!(list_queue(&mut app, &thread_id).await?.data.is_empty()); + Ok(()) +} + +#[tokio::test] +async fn a_new_turn_preserves_queued_messages_until_it_completes() -> Result<()> { + skip_if_remote!( + Ok(()), + "uses a host-local command and cwd fixture unavailable to remote executors" + ); + + let responses = vec![ + blocked_turn_response()?, + blocked_turn_response()?, + create_final_assistant_message_sse_response("new turn done")?, + create_final_assistant_message_sse_response("first queued message done")?, + create_final_assistant_message_sse_response("second queued message done")?, + ]; + let (mut app, _codex_home, _server) = queue_app(responses).await?; + let thread_id = app + .start_thread(ThreadStartParams::default()) + .await? + .thread + .id; + let (active_turn_id, _approval_id) = start_blocked_turn(&mut app, &thread_id).await?; + + let first = queue_item( + &mut app, + ThreadQueueAddParams { + client_user_message_id: "first-queued-client-id".to_string(), + ..submission(&thread_id, "first queued message") + }, + ) + .await?; + let second = queue_item( + &mut app, + ThreadQueueAddParams { + client_user_message_id: "second-queued-client-id".to_string(), + ..submission(&thread_id, "second queued message") + }, + ) + .await?; + + let _: TurnInterruptResponse = app + .request(|request_id| ClientRequest::TurnInterrupt { + request_id, + params: TurnInterruptParams { + thread_id: thread_id.clone(), + turn_id: active_turn_id, + }, + }) + .await?; + let interrupted: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(interrupted.turn.status, TurnStatus::Interrupted); + + let _: TurnStartResponse = app + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: thread_id.clone(), + input: first.input.clone(), + client_user_message_id: Some(first.client_user_message_id.clone()), + ..Default::default() + }, + }) + .await?; + let approval = timeout(READ_TIMEOUT, app.read_stream_until_request_message()).await??; + let ServerRequest::CommandExecutionRequestApproval { + request_id: new_approval_id, + .. + } = approval + else { + anyhow::bail!("matching ordinary turn did not request command approval"); + }; + assert_eq!( + vec![first, second], + list_queue(&mut app, &thread_id).await?.data + ); + + decline_approval(&mut app, new_approval_id).await?; + for _ in 0..3 { + let completed: TurnCompletedNotification = + timeout(READ_TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + } + assert!(list_queue(&mut app, &thread_id).await?.data.is_empty()); + + Ok(()) +} + +async fn queue_app(responses: Vec) -> Result<(TestAppServer, TempDir, MockServer)> { + let server = create_mock_responses_server_sequence(responses).await; + queue_app_with_server(server).await +} + +async fn queue_app_with_server(server: MockServer) -> Result<(TestAppServer, TempDir, MockServer)> { + let codex_home = TempDir::new()?; + let config = MockResponsesConfig::new(&server.uri()) + .with_approval_policy("untrusted") + .with_root_config(r#"approvals_reviewer = "user""#); + config.write(codex_home.path())?; + let app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized() + .await?; + Ok((app, codex_home, server)) +} + +fn blocked_turn_response() -> Result { + #[cfg(target_os = "windows")] + let shell_command = vec![ + "powershell".to_string(), + "-Command".to_string(), + "Start-Sleep -Seconds 10".to_string(), + ]; + #[cfg(not(target_os = "windows"))] + let shell_command = vec![ + "python3".to_string(), + "-c".to_string(), + "import time; time.sleep(10)".to_string(), + ]; + + create_shell_command_sse_response( + shell_command, + /*workdir*/ None, + /*timeout_ms*/ Some(10_000), + "queue-blocked-command", + ) +} + +async fn start_blocked_turn( + app: &mut TestAppServer, + thread_id: &str, +) -> Result<(String, RequestId)> { + let started: TurnStartResponse = app + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: thread_id.to_string(), + input: vec![text("start an approval-blocked turn")], + ..Default::default() + }, + }) + .await?; + let approval = timeout(READ_TIMEOUT, app.read_stream_until_request_message()).await??; + let ServerRequest::CommandExecutionRequestApproval { request_id, .. } = approval else { + anyhow::bail!("active turn did not request command approval"); + }; + Ok((started.turn.id, request_id)) +} + +async fn queue_item( + app: &mut TestAppServer, + params: ThreadQueueAddParams, +) -> Result { + let response: ThreadQueueAddResponse = app + .request(|request_id| ClientRequest::ThreadQueueAdd { request_id, params }) + .await?; + Ok(response.queued_submission) +} + +async fn list_queue(app: &mut TestAppServer, thread_id: &str) -> Result { + let mut data = Vec::new(); + let mut cursor = None; + loop { + let page: ThreadQueueListResponse = app + .request(|request_id| ClientRequest::ThreadQueueList { + request_id, + params: ThreadQueueListParams { + thread_id: thread_id.to_string(), + cursor, + limit: None, + }, + }) + .await?; + data.extend(page.data); + cursor = page.next_cursor; + if cursor.is_none() { + return Ok(ThreadQueueListResponse { + data, + next_cursor: None, + }); + } + } +} + +async fn decline_approval(app: &mut TestAppServer, request_id: RequestId) -> Result<()> { + app.send_response( + request_id, + serde_json::to_value(CommandExecutionRequestApprovalResponse { + decision: CommandExecutionApprovalDecision::Decline, + })?, + ) + .await +} + +fn submission(thread_id: &str, value: &str) -> ThreadQueueAddParams { + ThreadQueueAddParams { + thread_id: thread_id.to_string(), + input: vec![text(value)], + client_user_message_id: format!("queued-{value}"), + } +} + +fn text(value: &str) -> UserInput { + UserInput::Text { + text: value.to_string(), + text_elements: Vec::new(), + } +} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 2dc42fef6d..de3fe88441 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1168,6 +1168,14 @@ impl Session { self.services.elicitations.subscribe() } + pub(crate) fn mark_interrupted(&self) { + self.agent_status.send_replace(AgentStatus::Interrupted); + } + + pub(crate) fn is_interrupted(&self) -> bool { + matches!(*self.agent_status.borrow(), AgentStatus::Interrupted) + } + pub(crate) fn get_tx_event(&self) -> Sender { self.tx_event.clone() } diff --git a/codex-rs/core/src/tasks/lifecycle.rs b/codex-rs/core/src/tasks/lifecycle.rs index 0af3127d95..cde3fc7d72 100644 --- a/codex-rs/core/src/tasks/lifecycle.rs +++ b/codex-rs/core/src/tasks/lifecycle.rs @@ -41,9 +41,18 @@ impl Session { } pub(crate) async fn emit_thread_idle_lifecycle_if_idle(&self, cause: ThreadIdleCause) { - if self.active_turn.lock().await.is_some() - || self.input_queue.has_trigger_turn_mailbox_items().await - { + let cause = { + let active_turn = self.active_turn.lock().await; + if active_turn.is_some() { + return; + } + if self.is_interrupted() { + ThreadIdleCause::Interrupted + } else { + cause + } + }; + if self.input_queue.has_trigger_turn_mailbox_items().await { return; } diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 3f1c7f9c0f..cc71acb3b5 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -495,7 +495,7 @@ impl Session { let mut aborted_turn = false; let mut active_turn_to_clear = None; let mut turn_context = None; - if let Some(mut active_turn) = self.take_active_turn().await { + if let Some(mut active_turn) = self.take_active_turn(&reason).await { let task = active_turn.task.take(); aborted_turn = task.is_some(); turn_context = task.as_ref().map(|task| Arc::clone(&task.turn_context)); @@ -533,6 +533,12 @@ impl Session { .and_then(|active_turn| active_turn.task.as_ref()) .is_some_and(|task| task.turn_context.sub_id == turn_id) { + if matches!( + reason, + TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited + ) { + self.mark_interrupted(); + } active.take() } else { None @@ -769,7 +775,10 @@ impl Session { turn_id: turn_context.sub_id.clone(), profile, }); - let idle_cause = if matches!(abort_reason.as_ref(), Some(TurnAbortReason::Interrupted)) { + let idle_cause = if matches!( + abort_reason.as_ref(), + Some(TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited) + ) { ThreadIdleCause::Interrupted } else if abort_reason.is_none() && turn_context.terminal_error.lock().await.is_some() { ThreadIdleCause::Failed @@ -836,8 +845,17 @@ impl Session { } } - async fn take_active_turn(&self) -> Option { + async fn take_active_turn(&self, reason: &TurnAbortReason) -> Option { let mut active = self.active_turn.lock().await; + if matches!( + reason, + TurnAbortReason::Interrupted | TurnAbortReason::BudgetLimited + ) && active + .as_ref() + .is_some_and(|active_turn| active_turn.task.is_some()) + { + self.mark_interrupted(); + } active.take() } diff --git a/codex-rs/ext/queue/src/service.rs b/codex-rs/ext/queue/src/service.rs index 6f1e049e46..f0a1727703 100644 --- a/codex-rs/ext/queue/src/service.rs +++ b/codex-rs/ext/queue/src/service.rs @@ -4,7 +4,6 @@ use std::sync::Mutex as StdMutex; use std::sync::Weak; use codex_core::CodexThread; -use codex_core::NotSubmittedReason; use codex_core::StartIfIdleSubmission; use codex_core::ThreadManager; use codex_core::TurnInput; @@ -44,14 +43,12 @@ pub struct QueuedItem { pub enum QueueServiceError { #[error("queue storage failed: {0}")] Storage(#[from] ThreadStoreError), - #[error("queued item payload is invalid: {0}")] + #[error("queued submission payload is invalid: {0}")] InvalidPayload(#[from] serde_json::Error), #[error("local queued attachment is invalid: {0}")] InvalidAttachment(#[from] std::io::Error), #[error("Core failed to submit queued user message: {0}")] CoreSubmissionError(#[from] CodexErr), - #[error("Core declined to start queued user message: {0:?}")] - NotStarted(NotSubmittedReason), #[error("only user input can be added to the user-message queue")] InvalidInput, #[error( @@ -107,15 +104,29 @@ impl QueuedItemService { ) -> Result { let input = prepare_queued_user_input(input).await?; let payload = serde_json::to_string(&input)?; - let item = queued_item_from_record(self.queue.enqueue(thread_id, payload).await?)?; - self.emit_changed(thread_id); + let item = { + let _dispatch_guard = self.dispatch_guard(thread_id).await; + let item = queued_item_from_record(self.queue.enqueue(thread_id, payload).await?)?; + self.emit_changed(thread_id); + item + }; self.wake_if_loaded(thread_id).await; Ok(item) } pub async fn list(&self, thread_id: ThreadId) -> Result, QueueServiceError> { + self.list_page(thread_id, /*offset*/ 0, MAX_QUEUE_ITEMS) + .await + } + + pub async fn list_page( + &self, + thread_id: ThreadId, + offset: usize, + limit: usize, + ) -> Result, QueueServiceError> { self.queue - .list_page(thread_id, /*offset*/ 0, MAX_QUEUE_ITEMS) + .list_page(thread_id, offset, limit) .await? .into_iter() .map(queued_item_from_record) @@ -129,9 +140,7 @@ impl QueuedItemService { mut input: TurnInput, ) -> Result, QueueServiceError> { let _dispatch_guard = self.dispatch_guard(thread_id).await; - if let TurnInput::UserInput { client_id, .. } = &mut input - && client_id.is_none() - { + if let TurnInput::UserInput { client_id, .. } = &mut input { *client_id = self .list(thread_id) .await? @@ -190,11 +199,11 @@ impl QueuedItemService { Ok(()) } - /// Starts the exact queued message when idle and removes it after submission. + /// Starts the selected queued message only when its thread is idle. pub async fn start( &self, thread: &CodexThread, - queued_item_id: String, + queued_item_id: Option, trace: Option, ) -> Result { let thread_id = thread.session_configured().thread_id; @@ -203,21 +212,23 @@ impl QueuedItemService { .list(thread_id) .await? .into_iter() - .find(|item| item.id == queued_item_id) + .find(|item| queued_item_id.as_ref().is_none_or(|id| item.id == *id)) .ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("queued item not found: {queued_item_id}"), + message: queued_item_id.as_ref().map_or_else( + || "queue is empty".to_string(), + |id| format!("queued submission not found: {id}"), + ), })?; + let queued_item_id = item.id.clone(); let input @ TurnInput::UserInput { .. } = item.input else { return Err(QueueServiceError::InvalidInput); }; let submission = thread .start_turn_if_idle(TurnInputRequest::new(input).with_trace(trace)) - .await - .map_err(QueueServiceError::CoreSubmissionError)?; - if let StartIfIdleSubmission::NotSubmitted { reason } = submission { - return Err(QueueServiceError::NotStarted(reason)); + .await?; + if matches!(submission, StartIfIdleSubmission::Started { .. }) { + self.delete_locked(thread_id, queued_item_id).await?; } - self.delete_locked(thread_id, queued_item_id).await?; Ok(submission) } @@ -307,10 +318,22 @@ impl QueuedItemService { } async fn prepare_queued_user_input(mut input: TurnInput) -> Result { - validate_queued_user_input(&input)?; let TurnInput::UserInput { content, client_id } = &mut input else { return Err(QueueServiceError::InvalidInput); }; + if content.is_empty() { + return Err(QueueServiceError::InvalidInput); + } + let actual_chars: usize = content + .iter() + .filter_map(|item| match item { + UserInput::Text { text, .. } => Some(text.chars().count()), + _ => None, + }) + .sum(); + if actual_chars > MAX_USER_INPUT_TEXT_CHARS { + return Err(QueueServiceError::InputTooLarge { actual_chars }); + } client_id.get_or_insert_with(|| Uuid::now_v7().to_string()); if !content.iter().any(|item| { matches!( @@ -335,33 +358,13 @@ async fn prepare_queued_user_input(mut input: TurnInput) -> Result Result<(), QueueServiceError> { - let TurnInput::UserInput { content, .. } = input else { - return Err(QueueServiceError::InvalidInput); - }; - if content.is_empty() { - return Err(QueueServiceError::InvalidInput); - } - let actual_chars: usize = content - .iter() - .filter_map(|item| match item { - UserInput::Text { text, .. } => Some(text.chars().count()), - _ => None, - }) - .sum(); - if actual_chars > MAX_USER_INPUT_TEXT_CHARS { - return Err(QueueServiceError::InputTooLarge { actual_chars }); - } - Ok(()) -} - impl ThreadLifecycleContributor for QueuedItemService where C: Send + Sync + 'static, { fn on_thread_idle<'a>(&'a self, input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> { Box::pin(async move { - if input.cause != ThreadIdleCause::Completed { + if input.cause == ThreadIdleCause::Interrupted { return; } let Ok(thread_id) = ThreadId::from_string(input.thread_store.level_id()) else { diff --git a/codex-rs/ext/queue/tests/queue_service.rs b/codex-rs/ext/queue/tests/queue_service.rs index 407bbcdaa4..db63540b4c 100644 --- a/codex-rs/ext/queue/tests/queue_service.rs +++ b/codex-rs/ext/queue/tests/queue_service.rs @@ -226,24 +226,26 @@ async fn queued_input_and_unique_event_ids_round_trip() -> anyhow::Result<()> { 7, uuid::Uuid::parse_str(generated_client_id)?.get_version_num() ); - assert_eq!(vec![first.clone(), second], service.list(thread_id).await?); + assert_eq!( + vec![first.clone(), second.clone()], + service.list(thread_id).await? + ); assert_eq!(input, first.input); - let events = sink - .events - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - assert_eq!(2, events.len()); - for event in events.iter() { - let EventMsg::ThreadQueueChanged(change) = &event.msg else { - anyhow::bail!("event is not a queue change"); - }; - assert_eq!( - serde_json::json!({ "threadId": thread_id }), - serde_json::to_value(change)? - ); + { + let events = sink + .events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert_eq!(2, events.len()); + for event in events.iter() { + let EventMsg::ThreadQueueChanged(change) = &event.msg else { + anyhow::bail!("event is not a queue change"); + }; + assert_eq!(change.thread_id, thread_id); + } + assert_ne!(events[0].id, events[1].id); } - assert_ne!(events[0].id, events[1].id); Ok(()) } @@ -298,10 +300,17 @@ async fn starting_a_selected_item_preserves_the_remaining_queue() -> anyhow::Res .await?; let submission = service - .start(test.codex.as_ref(), second.id.clone(), /*trace*/ None) + .start( + test.codex.as_ref(), + Some(second.id.clone()), + /*trace*/ None, + ) .await?; - assert!(matches!(submission, StartIfIdleSubmission::Started { .. })); + assert!(matches!( + submission, + StartIfIdleSubmission::Started { turn_id } if !turn_id.is_empty() + )); assert_eq!(vec![first], service.list(thread_id).await?); wait_for_event_match(test.codex.as_ref(), |event| match event { EventMsg::TurnComplete(_) => Some(()), @@ -358,13 +367,18 @@ async fn starting_a_selected_item_while_active_leaves_it_queued() -> anyhow::Res ) .await?; - let error = service - .start(test.codex.as_ref(), queued.id.clone(), /*trace*/ None) - .await - .expect_err("active turn should reject explicit queue start"); + let submission = service + .start( + test.codex.as_ref(), + Some(queued.id.clone()), + /*trace*/ None, + ) + .await?; assert!(matches!( - error, - QueueServiceError::NotStarted(NotSubmittedReason::NotIdle) + submission, + StartIfIdleSubmission::NotSubmitted { + reason: NotSubmittedReason::NotIdle + } )); assert_eq!(vec![queued], service.list(thread_id).await?); @@ -379,8 +393,11 @@ async fn starting_a_selected_item_while_active_leaves_it_queued() -> anyhow::Res } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn stopped_turns_leave_queued_messages_in_place() -> anyhow::Result<()> { +async fn interrupted_turns_pause_queued_messages_but_failed_turns_drain_them() -> anyhow::Result<()> +{ let server = start_mock_server().await; + let response = + responses::mount_sse_once(&server, responses::sse_completed("failed-follow-up")).await; let test = test_codex().build_with_auto_env(&server).await?; let thread_id = test.session_configured.thread_id; let queue = loaded_thread_queue(&test)?; @@ -390,14 +407,26 @@ async fn stopped_turns_leave_queued_messages_in_place() -> anyhow::Result<()> { Arc::new(NoopExtensionEventSink), ); let queued = service - .enqueue(thread_id, user_input("keep queued after stopping")) + .enqueue(thread_id, user_input("continue after failure")) .await?; - for cause in [ThreadIdleCause::Interrupted, ThreadIdleCause::Failed] { - emit_idle_with_cause(&service, thread_id, cause).await; + emit_idle_with_cause(&service, thread_id, ThreadIdleCause::Interrupted).await; + assert_eq!(vec![queued], service.list(thread_id).await?); - assert_eq!(vec![queued.clone()], service.list(thread_id).await?); - } + emit_idle_with_cause(&service, thread_id, ThreadIdleCause::Failed).await; + wait_for_event_match(test.codex.as_ref(), |event| { + matches!(event, EventMsg::TurnComplete(_)).then_some(()) + }) + .await; + assert!(service.list(thread_id).await?.is_empty()); + assert_eq!( + Some("continue after failure"), + response + .single_request() + .message_input_texts("user") + .last() + .map(String::as_str) + ); Ok(()) } @@ -541,7 +570,7 @@ async fn explicitly_started_rejected_queue_messages_are_consumed() -> anyhow::Re let rejected = queue.enqueue(thread_id, user_input("blocked")).await?; let submission = tokio::time::timeout( Duration::from_secs(10), - queue.start(test.codex.as_ref(), rejected.id, /*trace*/ None), + queue.start(test.codex.as_ref(), Some(rejected.id), /*trace*/ None), ) .await? .expect("explicitly started input should be submitted"); diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 1b0408c264..ca6d1ec1c1 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -58,8 +58,6 @@ use crate::request_user_input::RequestUserInputResponse; use crate::turn_input::TurnInputMode; use crate::turn_input::TurnInputRequest; use crate::turn_input::TurnInputSubmission; -#[cfg(test)] -use crate::user_input::UserInput; use codex_extension_items::image_generation::ImageGenerationFailure; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; @@ -3839,7 +3837,7 @@ pub struct ThreadGoalUpdatedEvent { pub goal: ThreadGoal, } -#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "protocol/")] pub struct ThreadQueueChangedEvent { @@ -5562,7 +5560,7 @@ mod tests { #[test] fn user_input_text_serializes_empty_text_elements() -> Result<()> { - let input = UserInput::Text { + let input = crate::user_input::UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }; diff --git a/codex-rs/state/src/runtime/queued_items.rs b/codex-rs/state/src/runtime/queued_items.rs index 571d5eaa94..85f0c1f016 100644 --- a/codex-rs/state/src/runtime/queued_items.rs +++ b/codex-rs/state/src/runtime/queued_items.rs @@ -122,7 +122,7 @@ impl SqliteQueueStore { transaction.rollback().await?; return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, - "queue reorder must include every queued item exactly once", + "queue reorder must include every queued submission exactly once", ) .into()); } diff --git a/codex-rs/thread-store/src/queue_store.rs b/codex-rs/thread-store/src/queue_store.rs index 6de7150b63..606afb1014 100644 --- a/codex-rs/thread-store/src/queue_store.rs +++ b/codex-rs/thread-store/src/queue_store.rs @@ -83,7 +83,9 @@ impl QueueStore for LocalQueueStore { .await .map_err(|error| match error.downcast_ref::() { Some(sqlx::Error::RowNotFound) => ThreadStoreError::InvalidRequest { - message: format!("queue cannot contain more than {MAX_QUEUE_ITEMS} items"), + message: format!( + "queue cannot contain more than {MAX_QUEUE_ITEMS} submissions" + ), }, _ => ThreadStoreError::Internal { message: format!("queue storage failed: {error}"), 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 c3d39f89d7..e67a386644 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -68,6 +68,9 @@ pub(super) fn server_notification_thread_target( ServerNotification::ThreadGoalCleared(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::ThreadQueueChanged(notification) => { + Some(notification.thread_id.as_str()) + } ServerNotification::ThreadSettingsUpdated(notification) => { Some(notification.thread_id.as_str()) } diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 69358321f1..5efc024bea 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -199,6 +199,7 @@ impl ChatWidget { | ServerNotification::ThreadStarted(_) | ServerNotification::ThreadStatusChanged(_) | ServerNotification::ThreadReverted(_) + | ServerNotification::ThreadQueueChanged(_) | ServerNotification::ThreadArchived(_) | ServerNotification::ThreadDeleted(_) | ServerNotification::ThreadUnarchived(_)