diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 778086152a..3d786fe1e5 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -5066,6 +5066,17 @@ ], "type": "object" }, + "ThreadRevertedNotification": { + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, "ThreadSection": { "description": "An independently persisted, user-visible thread section.", "properties": { @@ -6149,6 +6160,26 @@ "title": "Thread/closedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/reverted" + ], + "title": "Thread/revertedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRevertedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/revertedNotification", + "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 cc51aece19..e8735af196 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 @@ -4477,6 +4477,26 @@ "title": "Thread/closedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/reverted" + ], + "title": "Thread/revertedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadRevertedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/revertedNotification", + "type": "object" + }, { "properties": { "method": { @@ -20968,6 +20988,19 @@ "title": "ThreadResumeResponse", "type": "object" }, + "ThreadRevertedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRevertedNotification", + "type": "object" + }, "ThreadRollbackParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "DEPRECATED: `thread/rollback` will be removed soon.", 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 49b9eaeae9..fe54f8b240 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 @@ -14447,6 +14447,26 @@ "title": "Thread/closedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/reverted" + ], + "title": "Thread/revertedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadRevertedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/revertedNotification", + "type": "object" + }, { "properties": { "method": { @@ -18720,6 +18740,19 @@ "title": "ThreadResumeResponse", "type": "object" }, + "ThreadRevertedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRevertedNotification", + "type": "object" + }, "ThreadRollbackParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "DEPRECATED: `thread/rollback` will be removed soon.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRevertedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRevertedNotification.json new file mode 100644 index 0000000000..f768659960 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRevertedNotification.json @@ -0,0 +1,13 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadRevertedNotification", + "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 2a1084e87d..181acb6ff1 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 24b643f178..0d2b8a0122 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 271bfd4fa0..48fd6d2b0a 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -59,6 +59,7 @@ import type { ThreadRealtimeSdpNotification } from "./v2/ThreadRealtimeSdpNotifi import type { ThreadRealtimeStartedNotification } from "./v2/ThreadRealtimeStartedNotification"; import type { ThreadRealtimeTranscriptDeltaNotification } from "./v2/ThreadRealtimeTranscriptDeltaNotification"; import type { ThreadRealtimeTranscriptDoneNotification } from "./v2/ThreadRealtimeTranscriptDoneNotification"; +import type { ThreadRevertedNotification } from "./v2/ThreadRevertedNotification"; import type { ThreadSettingsUpdatedNotification } from "./v2/ThreadSettingsUpdatedNotification"; import type { ThreadStartedNotification } from "./v2/ThreadStartedNotification"; import type { ThreadStatusChangedNotification } from "./v2/ThreadStatusChangedNotification"; @@ -76,4 +77,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": "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/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 46c4fb068b..84f048e8ef 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts @@ -59,6 +59,7 @@ import type { ThreadRealtimeSdpNotification } from "./v2/ThreadRealtimeSdpNotifi import type { ThreadRealtimeStartedNotification } from "./v2/ThreadRealtimeStartedNotification"; import type { ThreadRealtimeTranscriptDeltaNotification } from "./v2/ThreadRealtimeTranscriptDeltaNotification"; import type { ThreadRealtimeTranscriptDoneNotification } from "./v2/ThreadRealtimeTranscriptDoneNotification"; +import type { ThreadRevertedNotification } from "./v2/ThreadRevertedNotification"; import type { ThreadSettingsUpdatedNotification } from "./v2/ThreadSettingsUpdatedNotification"; import type { ThreadStartedNotification } from "./v2/ThreadStartedNotification"; import type { ThreadStatusChangedNotification } from "./v2/ThreadStatusChangedNotification"; @@ -86,4 +87,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": "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/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/ThreadRevertedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRevertedNotification.ts new file mode 100644 index 0000000000..803d1cc7f9 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRevertedNotification.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 ThreadRevertedNotification = { 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 017c21615b..0671837f41 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -478,6 +478,7 @@ export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeT export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams"; export type { ThreadResumeParams } from "./ThreadResumeParams"; export type { ThreadResumeResponse } from "./ThreadResumeResponse"; +export type { ThreadRevertedNotification } from "./ThreadRevertedNotification"; export type { ThreadRollbackParams } from "./ThreadRollbackParams"; export type { ThreadRollbackResponse } from "./ThreadRollbackResponse"; export type { ThreadSearchResult } from "./ThreadSearchResult"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 6d0866a863..813b8b0065 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -646,6 +646,12 @@ client_request_definitions! { serialization: thread_id(params.thread_id), response: v2::ThreadRollbackResponse, }, + #[experimental("thread/revert")] + ThreadRevert => "thread/revert" { + params: v2::ThreadRevertParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadRevertResponse, + }, ThreadList => "thread/list" { params: v2::ThreadListParams, inspect_params: true, @@ -1711,6 +1717,8 @@ server_notification_definitions! { ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification), ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification), ThreadClosed => "thread/closed" (v2::ThreadClosedNotification), + #[experimental("thread/reverted")] + ThreadReverted => "thread/reverted" (v2::ThreadRevertedNotification), SkillsChanged => "skills/changed" (v2::SkillsChangedNotification), ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification), ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification), 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 fbf79341ca..5c8603c5b3 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -437,17 +437,17 @@ pub struct ThreadResumeResponse { #[experimental("thread/resume.initialTurnsPage")] #[serde(default)] pub initial_turns_page: Option, - /// Opaque head cursor for hydrating paginated turns backwards. + /// Opaque cursor for hydrating paginated turns backwards. /// /// Pass this as `cursor` to `thread/turns/list` with - /// `sortDirection: "desc"`. The first page includes the cursor's head turn. + /// `sortDirection: "desc"`. The first page includes the turn identified by the cursor. #[experimental("thread/resume.turnsBackwardsCursor")] #[serde(default)] pub turns_backwards_cursor: Option, - /// Opaque head cursor for hydrating paginated items backwards. + /// Opaque cursor for hydrating paginated items backwards. /// /// Pass this as `cursor` to `thread/items/list` with - /// `sortDirection: "desc"`. The first page includes the cursor's head item. + /// `sortDirection: "desc"`. The first page includes the item identified by the cursor. #[experimental("thread/resume.itemsBackwardsCursor")] #[serde(default)] pub items_backwards_cursor: Option, @@ -1117,6 +1117,37 @@ pub struct ThreadRollbackResponse { pub thread: Thread, } +/// Replace a paginated thread's durable history with the prefix before one turn. +/// +/// This only changes persisted conversation history. It does not revert local file changes. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRevertParams { + pub thread_id: String, + /// Turn excluded from the replacement history, together with every later turn. + pub before_turn_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRevertResponse { + /// Updated loaded thread metadata. `turns` is always empty; hydrate retained history through + /// `thread/turns/list`. + pub thread: Thread, + /// Opaque cursor for hydrating paginated turns backwards. + /// + /// Pass this as `cursor` to `thread/turns/list` with + /// `sortDirection: "desc"`. The first page includes the turn identified by the cursor. + pub turns_backwards_cursor: Option, + /// Opaque cursor for hydrating paginated items backwards. + /// + /// Pass this as `cursor` to `thread/items/list` with + /// `sortDirection: "desc"`. The first page includes the item identified by the cursor. + pub items_backwards_cursor: Option, +} + /// Parameters for listing independently persisted thread sections. #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] @@ -1704,6 +1735,13 @@ pub struct ThreadUnarchivedNotification { pub struct ThreadClosedNotification { pub thread_id: String, } + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadRevertedNotification { + pub thread_id: String, +} #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 476b1871e2..a99f9adf3a 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -198,6 +198,7 @@ Example with notification opt-out: - `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids. - `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated. - `thread/rollback` — deprecated and will be removed soon. Drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. Paginated threads do not support rollback. +- `thread/revert` — experimental. Replace a loaded paginated thread's durable history with the prefix strictly before `beforeTurnId` while preserving its thread id. The operation interrupts an active turn if needed, leaves older rollout files immutable, reloads the thread, returns updated thread metadata with empty `turns` plus pagination cursors, and emits `thread/reverted`. It does not revert local file changes. - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` supplies the default roots for newly resolved environment selections. Explicit `environments[].runtimeWorkspaceRoots` override that fallback with environment-native absolute paths. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". Deprecated experimental `multiAgentMode` is ignored; Ultra reasoning effort selects proactive behavior. - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success. - `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Review and manual compaction turns reject `turn/steer`. @@ -352,7 +353,7 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. A cold paginated resume can still replay persisted `thread/tokenUsage/updated` when it can identify the corresponding stored turn; resuming an already-loaded thread waits for the next live update. -Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. +Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the row identified by the cursor, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. Only one app-server process can hold a paginated thread open for writing at a time. If another process already owns the thread, `thread/resume`, `thread/archive`, and `thread/delete` fail with JSON-RPC error `-32600`. Archive and deletion also fail if another process owns any spawned descendant. Read-only requests remain available without resuming the thread. @@ -375,8 +376,8 @@ Example: } } { "id": 12, "result": { "thread": { "id": "thr_123", "turns": [], … }, - "turnsBackwardsCursor": "turn-head-cursor-or-null", - "itemsBackwardsCursor": "item-head-cursor-or-null" + "turnsBackwardsCursor": "turn-backwards-cursor-or-null", + "itemsBackwardsCursor": "item-backwards-cursor-or-null" } } { "method": "thread/resume", "id": 13, "params": { diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index a9e57a3e3e..0d7480b923 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1181,6 +1181,16 @@ impl MessageProcessor { .thread_rollback(&request_id, params, app_server_client_name.as_deref()) .await } + ClientRequest::ThreadRevert { params, .. } => { + self.thread_processor + .thread_revert( + request_id.clone(), + params, + app_server_client_name.clone(), + client_version.clone(), + ) + .await + } ClientRequest::ThreadList { params, .. } => { self.thread_processor.thread_list(params).await } 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 1119f1c56b..34ecd65b48 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -346,6 +346,14 @@ pub(super) async fn ensure_listener_task_running( fallback_model_provider.clone(), ) .await; + if matches!(event.msg, EventMsg::ShutdownComplete) + && let Some(completion_tx) = thread_state + .lock() + .await + .take_shutdown_drain_waiter() + { + let _ = completion_tx.send(()); + } } unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => { if !unloading_watchers_open { @@ -421,11 +429,14 @@ pub(super) async fn unload_thread_without_subscribers( tokio::spawn(async move { match wait_for_thread_shutdown(&thread).await { ThreadShutdownResult::Complete => { - if thread_manager.remove_thread(&thread_id).await.is_none() { - info!("thread {thread_id} was already removed before teardown finalized"); - thread_watch_manager - .remove_thread(&thread_id.to_string()) - .await; + // A delayed unload can finish after thread/revert replaces this runtime under + // the same thread ID. Only the runtime that scheduled this unload may remove it. + if thread_manager + .remove_thread_if_matches(&thread_id, &thread) + .await + .is_none() + { + info!("thread {thread_id} was replaced or removed before teardown finalized"); pending_thread_unloads.lock().await.remove(&thread_id); return; } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index b1eb05eab6..06f670b56e 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -4,6 +4,9 @@ use super::turn_processor::can_accept_direct_input; use super::*; use crate::error_code::method_not_found; use codex_app_server_protocol::SelectedCapabilityRoot; +use codex_app_server_protocol::ThreadRevertParams; +use codex_app_server_protocol::ThreadRevertResponse; +use codex_app_server_protocol::ThreadRevertedNotification; use codex_app_server_protocol::ThreadSection; use codex_app_server_protocol::ThreadSectionAppearance; use codex_app_server_protocol::ThreadSectionMoveParams; @@ -33,6 +36,12 @@ struct ThreadListFilters { relation_filter: Option, } +struct ThreadRevertRuntimeSnapshot { + config: Config, + settings: ThreadConfigSnapshot, + client_mcp_extensions: ClientMcpExtensions, +} + fn collect_resume_override_mismatches( request: &ThreadResumeParams, config_snapshot: &ThreadConfigSnapshot, @@ -554,6 +563,30 @@ impl ThreadRequestProcessor { .map(|()| None) } + pub(crate) async fn thread_revert( + &self, + request_id: ConnectionRequestId, + params: ThreadRevertParams, + app_server_client_name: Option, + app_server_client_version: Option, + ) -> Result, JSONRPCErrorError> { + let (response, thread_id) = self + .thread_revert_response( + &request_id, + params, + app_server_client_name, + app_server_client_version, + ) + .await?; + self.outgoing.send_response(request_id, response).await; + self.outgoing + .send_server_notification(ServerNotification::ThreadReverted( + ThreadRevertedNotification { thread_id }, + )) + .await; + Ok(None) + } + pub(crate) async fn thread_archive( &self, request_id: ConnectionRequestId, @@ -1532,7 +1565,7 @@ impl ThreadRequestProcessor { archive_thread_ids.push(thread_id); } } - Err(err) => return Err(thread_store_archive_error("archive", err)), + Err(err) => return Err(thread_store_mutation_error("archive", err)), } for descendant_thread_id in subtree_thread_ids.iter().copied().skip(1) { match self @@ -1573,7 +1606,7 @@ impl ThreadRequestProcessor { writer_lock_thread_ids: subtree_thread_ids, }) .await - .map_err(|err| thread_store_archive_error("archive", err))? + .map_err(|err| thread_store_mutation_error("archive", err))? .into_iter() .map(|thread_id| thread_id.to_string()) .collect(); @@ -1828,7 +1861,7 @@ impl ThreadRequestProcessor { .thread_store .unarchive_thread(StoreArchiveThreadParams { thread_id }) .await - .map_err(|err| thread_store_archive_error("unarchive", err))?; + .map_err(|err| thread_store_mutation_error("unarchive", err))?; let (mut thread, _) = thread_from_stored_thread(stored_thread, fallback_provider.as_str(), &self.config.cwd); @@ -1851,6 +1884,227 @@ impl ThreadRequestProcessor { self.thread_rollback_start(request_id, params).await } + async fn thread_revert_response( + &self, + request_id: &ConnectionRequestId, + params: ThreadRevertParams, + app_server_client_name: Option, + app_server_client_version: Option, + ) -> Result<(ThreadRevertResponse, String), JSONRPCErrorError> { + let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; + let ThreadRevertParams { + thread_id, + before_turn_id, + } = params; + let (thread_id, thread) = self.load_thread(&thread_id).await?; + let config_snapshot = thread.config_snapshot().await; + if !matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) { + return Err(invalid_request( + "thread/revert only supports paginated threads", + )); + } + let runtime_snapshot = ThreadRevertRuntimeSnapshot { + config: thread.config().await.as_ref().clone(), + settings: config_snapshot, + client_mcp_extensions: thread.client_mcp_extensions(), + }; + + // Subscribe before shutdown so a pending idle unload either rejects this request or can + // no longer race the replacement runtime. The same listener then drains Core's shutdown + // events before we replace it. + if matches!( + self.ensure_conversation_listener( + thread_id, + request_id.connection_id, + /*raw_events_enabled*/ false, + ) + .await?, + EnsureConversationListenerResult::ConnectionClosed + ) { + return Err(internal_error(format!( + "connection closed before thread {thread_id} could be reverted" + ))); + } + let thread_state = self.thread_state_manager.thread_state(thread_id).await; + let shutdown_drain_rx = thread_state.lock().await.register_shutdown_drain_waiter(); + + match wait_for_thread_shutdown(&thread).await { + ThreadShutdownResult::Complete => {} + ThreadShutdownResult::SubmitFailed => { + thread_state.lock().await.take_shutdown_drain_waiter(); + return Err(internal_error(format!( + "failed to shut down thread {thread_id} before revert" + ))); + } + ThreadShutdownResult::TimedOut => { + thread_state.lock().await.take_shutdown_drain_waiter(); + return Err(internal_error(format!( + "timed out shutting down thread {thread_id} before revert" + ))); + } + } + let drain_result = tokio::time::timeout(Duration::from_secs(10), shutdown_drain_rx) + .await + .map_err(|_| { + internal_error(format!( + "timed out waiting for thread {thread_id} listener to drain shutdown events" + )) + }) + .and_then(|result| { + result.map_err(|_| { + internal_error(format!( + "thread {thread_id} listener stopped before draining shutdown events" + )) + }) + }); + if let Err(err) = drain_result { + thread_state.lock().await.take_shutdown_drain_waiter(); + return Err(err); + } + if self + .thread_manager + .remove_thread(&thread_id) + .await + .is_none() + { + return Err(internal_error(format!( + "thread {thread_id} disappeared before revert" + ))); + } + // Keep thread state and subscriptions across the internal reload. Full teardown would + // force clients to call thread/resume after a successful revert. + self.outgoing + .cancel_requests_for_thread(thread_id, /*error*/ None) + .await; + + let revert_result = self + .thread_store + .revert_thread(codex_thread_store::RevertThreadParams { + thread_id, + before_turn_id, + }) + .await + .map_err(|err| thread_store_mutation_error("revert", err)); + let response = self + .reload_paginated_thread( + request_id, + thread_id, + runtime_snapshot, + app_server_client_name, + app_server_client_version, + ) + .await?; + revert_result?; + Ok((response, thread_id.to_string())) + } + + async fn reload_paginated_thread( + &self, + request_id: &ConnectionRequestId, + thread_id: ThreadId, + runtime_snapshot: ThreadRevertRuntimeSnapshot, + app_server_client_name: Option, + app_server_client_version: Option, + ) -> Result { + let ThreadRevertRuntimeSnapshot { + config, + settings, + client_mcp_extensions, + } = runtime_snapshot; + let thread_id_string = thread_id.to_string(); + let stored_thread = self + .read_stored_thread_for_resume( + thread_id_string.as_str(), + /*path*/ None, + /*include_history*/ false, + ) + .await?; + let (thread_history, resume_source_thread) = self + .load_resume_initial_history_from_stored_thread(stored_thread) + .await?; + let response_history = thread_history.clone(); + let NewThread { + thread_id: resumed_thread_id, + thread: codex_thread, + session_configured, + .. + } = self + .thread_manager + .resume_thread_with_history( + config, + thread_history, + self.auth_manager.clone(), + self.request_trace_context(request_id).await, + client_mcp_extensions, + ) + .await + .map_err(|err| internal_error(format!("error reloading thread after revert: {err}")))?; + if resumed_thread_id != thread_id { + return Err(internal_error(format!( + "thread {thread_id} reloaded as {resumed_thread_id} after revert" + ))); + } + codex_thread + .restore_thread_settings(settings) + .await + .map_err(|err| { + internal_error(format!( + "failed to restore thread settings after revert: {err}" + )) + })?; + Self::set_app_server_client_info( + codex_thread.as_ref(), + app_server_client_name, + app_server_client_version, + ) + .await?; + let SessionConfiguredEvent { rollout_path, .. } = session_configured; + let rollout_path = rollout_path.ok_or_else(|| { + internal_error(format!( + "rollout path missing after reloading thread {thread_id}" + )) + })?; + log_listener_attach_result( + self.ensure_conversation_listener( + thread_id, + request_id.connection_id, + /*raw_events_enabled*/ false, + ) + .await, + thread_id, + request_id.connection_id, + "thread", + ); + let mut thread = self + .load_thread_from_resume_source_or_send_internal( + thread_id, + codex_thread.as_ref(), + &response_history, + rollout_path.as_path(), + Some(resume_source_thread), + /*include_turns*/ false, + ) + .await + .map_err(internal_error)?; + self.thread_watch_manager.upsert_thread(&thread.id).await; + let thread_status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; + set_thread_status_and_interrupt_stale_turns( + &mut thread, + thread_status, + /*has_live_in_progress_turn*/ false, + ); + let (turns_backwards_cursor, items_backwards_cursor) = + Self::paginated_resume_backwards_cursors(self.thread_store.as_ref(), thread_id).await?; + Ok(ThreadRevertResponse { + thread, + turns_backwards_cursor, + items_backwards_cursor, + }) + } + async fn thread_rollback_start( &self, request_id: &ConnectionRequestId, @@ -2857,7 +3111,7 @@ impl ThreadRequestProcessor { // durable next cursor still starts after the last returned stored turn. let page_size = thread_turns_page_size(params.limit); if page_size == 1 { - // ThreadStore does not accept an empty page. Use its head cursor as + // ThreadStore does not accept an empty page. Use its backwards cursor as // the next cursor so the omitted durable row is returned next. let mut page = self .paginated_resume_initial_turns_page(thread_id, params) @@ -3883,6 +4137,32 @@ impl ThreadRequestProcessor { }; let stored_thread = result.map_err(thread_store_resume_read_error)?; + if let Some(requested_path) = path + && matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) + { + let current_thread = self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id: stored_thread.thread_id, + include_archived: true, + include_history: false, + }) + .await + .map_err(thread_store_resume_read_error)?; + if let Some(current_path) = current_thread.rollout_path.as_ref() + && !path_utils::paths_match_after_normalization( + codex_rollout::plain_rollout_path(requested_path).as_path(), + codex_rollout::plain_rollout_path(current_path).as_path(), + ) + { + return Err(invalid_request(format!( + "cannot resume paginated thread {} with stale path: requested {}, current {}; omit path and resume by thread id", + stored_thread.thread_id, + requested_path.display(), + current_path.display() + ))); + } + } if stored_thread.archived_at.is_some() { let thread_id = stored_thread.thread_id; return Err(invalid_request(format!( @@ -5188,7 +5468,7 @@ pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPC } } -fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { +fn thread_store_mutation_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { ThreadStoreError::InvalidRequest { message } | ThreadStoreError::Conflict { message } => { invalid_request(message) diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 3cd45f29ce..1e82d43777 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -94,6 +94,9 @@ pub(crate) struct ThreadState { pub(crate) pending_rollbacks: Option, pub(crate) turn_summary: TurnSummary, pub(crate) last_terminal_turn_id: Option, + /// Lets an internal runtime replacement wait until the old listener has processed Core's + /// `ShutdownComplete` event before that listener is superseded. + shutdown_drain_waiter: Option>, pub(crate) cancel_tx: Option>, pub(crate) experimental_raw_events: bool, pub(crate) listener_generation: u64, @@ -135,6 +138,7 @@ impl ThreadState { if let Some(cancel_tx) = self.cancel_tx.take() { let _ = cancel_tx.send(()); } + self.shutdown_drain_waiter = None; self.listener_command_tx = None; self.current_turn_history.reset(); self.listener_thread = None; @@ -155,6 +159,16 @@ impl ThreadState { self.current_turn_history.active_turn_snapshot() } + pub(crate) fn register_shutdown_drain_waiter(&mut self) -> oneshot::Receiver<()> { + let (completion_tx, completion_rx) = oneshot::channel(); + self.shutdown_drain_waiter = Some(completion_tx); + completion_rx + } + + pub(crate) fn take_shutdown_drain_waiter(&mut self) -> Option> { + self.shutdown_drain_waiter.take() + } + pub(crate) fn track_current_turn_event(&mut self, event_turn_id: &str, event: &EventMsg) { if let EventMsg::TurnStarted(payload) = event { self.turn_summary.started_at = payload.started_at; diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 6fa4a33797..84149012d2 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -92,6 +92,7 @@ mod thread_metadata_update; mod thread_name_websocket; mod thread_read; mod thread_resume; +mod thread_revert; mod thread_rollback; mod thread_sections; mod thread_settings_update; diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs new file mode 100644 index 0000000000..7287d2b2bf --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -0,0 +1,379 @@ +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_repeating_assistant; +use app_test_support::create_mock_responses_server_sequence; +use app_test_support::create_request_user_input_sse_response; +use codex_app_server_protocol::AskForApproval; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SortDirection; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadItemsListParams; +use codex_app_server_protocol::ThreadItemsListResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadRevertParams; +use codex_app_server_protocol::ThreadRevertResponse; +use codex_app_server_protocol::ThreadRevertedNotification; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadTurnsListParams; +use codex_app_server_protocol::ThreadTurnsListResponse; +use codex_app_server_protocol::TurnCompletedNotification; +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 codex_protocol::config_types::CollaborationMode; +use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Settings; +use codex_protocol::openai_models::ReasoningEffort; +use pretty_assertions::assert_eq; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + + let ThreadStartResponse { thread, .. } = mcp + .start_thread(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let stale_rollout_path = thread.path.clone().expect("thread rollout path"); + let mut turn_ids = Vec::new(); + for text in ["first", "second"] { + let completed = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + turn_ids.push(completed.turn.id); + } + + let ThreadRevertResponse { + thread: reverted_thread, + turns_backwards_cursor, + items_backwards_cursor, + } = mcp + .request(|request_id| ClientRequest::ThreadRevert { + request_id, + params: ThreadRevertParams { + thread_id: thread.id.clone(), + before_turn_id: turn_ids[1].clone(), + }, + }) + .await?; + let reverted: ThreadRevertedNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_notification("thread/reverted"), + ) + .await??; + assert_eq!(reverted.thread_id, thread.id); + + assert_eq!(reverted_thread.id, thread.id); + assert!(reverted_thread.turns.is_empty()); + assert!(items_backwards_cursor.is_some()); + assert_eq!( + turn_ids_from_cursor( + &mut mcp, + &thread.id, + turns_backwards_cursor, + /*sort_direction*/ None, + ) + .await?, + turn_ids[..1] + ); + let ThreadItemsListResponse { + data: reverted_items, + .. + } = mcp + .request(|request_id| ClientRequest::ThreadItemsList { + request_id, + params: ThreadItemsListParams { + thread_id: thread.id.clone(), + turn_id: None, + cursor: items_backwards_cursor, + limit: None, + sort_direction: None, + }, + }) + .await?; + assert!(!reverted_items.is_empty()); + assert!( + reverted_items + .iter() + .all(|item| item.turn_id == turn_ids[0]) + ); + + mcp.shutdown_gracefully().await?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + let stale_resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + path: Some(stale_rollout_path), + ..Default::default() + }) + .await?; + let stale_resume_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(stale_resume_id)), + ) + .await??; + assert!( + stale_resume_error.error.message.contains("stale path") + && stale_resume_error + .error + .message + .contains("omit path and resume by thread id"), + "unexpected resume error: {}", + stale_resume_error.error.message, + ); + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }) + .await?; + let _: ThreadResumeResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??; + let invalid_revert_id = mcp + .send_raw_request( + "thread/revert", + Some(serde_json::to_value(ThreadRevertParams { + thread_id: thread.id.clone(), + before_turn_id: "missing-turn".to_string(), + })?), + ) + .await?; + let invalid_revert_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(invalid_revert_id)), + ) + .await??; + assert_eq!( + invalid_revert_error.error.message, + "turn not found: missing-turn" + ); + + let third_turn = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "third".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let requests = server.received_requests().await.expect("response requests"); + let model_input = requests + .iter() + .rev() + .find(|request| request.url.path().ends_with("/responses")) + .expect("third turn response request") + .body_json::()?["input"] + .clone(); + let model_input = serde_json::to_string(&model_input)?; + assert!(model_input.contains("first")); + assert!(!model_input.contains("second")); + assert!(model_input.contains("third")); + assert_eq!( + turn_ids_from_cursor( + &mut mcp, + &thread.id, + /*cursor*/ None, + Some(SortDirection::Asc), + ) + .await?, + vec![turn_ids[0].clone(), third_turn.turn.id] + ); + Ok(()) +} + +#[tokio::test] +async fn thread_revert_interrupts_active_turn_and_keeps_thread_loaded() -> Result<()> { + let home = TempDir::new()?; + let server = create_mock_responses_server_sequence(vec![ + create_final_assistant_message_sse_response("first")?, + create_request_user_input_sse_response("call_blocked")?, + create_final_assistant_message_sse_response("third")?, + ]) + .await; + MockResponsesConfig::new(&server.uri()).write(home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + + let ThreadStartResponse { thread, .. } = mcp + .start_thread(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let first_turn = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "first".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + + let TurnStartResponse { turn: active_turn } = mcp + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "sleep".to_string(), + text_elements: Vec::new(), + }], + collaboration_mode: Some(CollaborationMode { + mode: ModeKind::Plan, + settings: Settings { + model: "mock-model".to_string(), + reasoning_effort: Some(ReasoningEffort::Medium), + developer_instructions: None, + }, + }), + approval_policy: Some(AskForApproval::Never), + ..Default::default() + }, + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + + let ThreadRevertResponse { + thread: reverted_thread, + turns_backwards_cursor, + items_backwards_cursor, + } = mcp + .request(|request_id| ClientRequest::ThreadRevert { + request_id, + params: ThreadRevertParams { + thread_id: thread.id.clone(), + before_turn_id: active_turn.id.clone(), + }, + }) + .await?; + let completed: TurnCompletedNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_notification("turn/completed"), + ) + .await??; + assert_eq!(completed.thread_id, thread.id); + assert_eq!(completed.turn.status, TurnStatus::Interrupted); + assert!(reverted_thread.turns.is_empty()); + assert!(items_backwards_cursor.is_some()); + assert_eq!( + turn_ids_from_cursor( + &mut mcp, + &thread.id, + turns_backwards_cursor, + /*sort_direction*/ None, + ) + .await?, + vec![first_turn.turn.id] + ); + + let resumed: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }, + }) + .await?; + assert_eq!(resumed.approval_policy, AskForApproval::Never); + + mcp.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id, + input: vec![UserInput::Text { + text: "third".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + Ok(()) +} + +async fn turn_ids_from_cursor( + mcp: &mut TestAppServer, + thread_id: &str, + cursor: Option, + sort_direction: Option, +) -> Result> { + let ThreadTurnsListResponse { data, .. } = mcp + .request(|request_id| ClientRequest::ThreadTurnsList { + request_id, + params: ThreadTurnsListParams { + thread_id: thread_id.to_string(), + cursor, + limit: None, + sort_direction, + items_view: None, + }, + }) + .await?; + Ok(data.into_iter().map(|turn| turn.id).collect()) +} + +async fn initialize_experimental(mcp: &mut TestAppServer) -> Result<()> { + let initialized = timeout( + DEFAULT_READ_TIMEOUT, + mcp.initialize_with_capabilities( + ClientInfo { + name: "test-client".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + Some(InitializeCapabilities { + experimental_api: true, + request_attestation: false, + opt_out_notification_methods: None, + mcp_server_openai_form_elicitation: false, + extensions: None, + }), + ), + ) + .await??; + assert!(matches!(initialized, JSONRPCMessage::Response(_))); + Ok(()) +} diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 1fd83545c5..cbfc251848 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -20,6 +20,7 @@ use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; use codex_protocol::mcp::CallToolResult; +use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::models::ActivePermissionProfile; use codex_protocol::models::ContentItem; use codex_protocol::models::PermissionProfile; @@ -125,6 +126,22 @@ impl ThreadConfigSnapshot { collaboration_mode: self.collaboration_mode, } } + + fn into_thread_settings_overrides(self) -> CodexThreadSettingsOverrides { + CodexThreadSettingsOverrides { + environments: Some(self.environments), + profile_workspace_roots: Some(self.profile_workspace_roots), + approval_policy: Some(self.approval_policy), + approvals_reviewer: Some(self.approvals_reviewer), + permission_profile: Some(self.permission_profile), + active_permission_profile: self.active_permission_profile, + summary: self.reasoning_summary, + service_tier: Some(self.service_tier), + collaboration_mode: Some(self.collaboration_mode), + personality: self.personality, + ..Default::default() + } + } } /// Thread settings overrides that app-server validates before starting a turn. @@ -406,6 +423,20 @@ impl CodexThread { self.session.preview_settings(&updates).await } + /// Restores effective mutable settings captured from another loaded runtime. + /// + /// Runtime replacement uses this after resume so clients keep their current thread settings + /// rather than reverting to the original layer-backed config. + pub async fn restore_thread_settings( + &self, + snapshot: ThreadConfigSnapshot, + ) -> ConstraintResult<()> { + let updates = self + .thread_settings_update(snapshot.into_thread_settings_overrides()) + .await; + self.session.update_settings(updates).await + } + async fn thread_settings_update( &self, overrides: CodexThreadSettingsOverrides, @@ -617,6 +648,11 @@ impl CodexThread { self.session.thread_config_snapshot().await } + /// Returns the MCP extensions declared by the client that created this runtime. + pub fn client_mcp_extensions(&self) -> ClientMcpExtensions { + self.session.services.client_mcp_extensions.clone() + } + /// Returns the files that supplied the thread's loaded model instructions. pub async fn instruction_sources(&self) -> Vec { self.session.instruction_sources().await diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 3e93007bab..42b14e3400 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1040,6 +1040,26 @@ impl ThreadManager { self.state.threads.write().await.remove(thread_id) } + /// Removes a thread only if `thread_id` still maps to `expected`. + /// + /// Delayed cleanup uses this to avoid removing a replacement runtime registered under the + /// same thread ID. + pub async fn remove_thread_if_matches( + &self, + thread_id: &ThreadId, + expected: &Arc, + ) -> Option> { + let mut threads = self.state.threads.write().await; + if threads + .get(thread_id) + .is_some_and(|thread| Arc::ptr_eq(thread, expected)) + { + threads.remove(thread_id) + } else { + None + } + } + /// Tries to shut down all tracked threads concurrently within the provided timeout. /// Threads that complete shutdown are removed from the manager; incomplete shutdowns /// remain tracked so callers can retry or inspect them later. 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 5935e46d68..c3d39f89d7 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -51,6 +51,7 @@ pub(super) fn server_notification_thread_target( ServerNotification::ThreadStatusChanged(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::ThreadReverted(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadArchived(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadDeleted(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadUnarchived(notification) => Some(notification.thread_id.as_str()), diff --git a/codex-rs/tui/src/app/tests/safety_buffering.rs b/codex-rs/tui/src/app/tests/safety_buffering.rs index 42fc6bce46..33ca6be628 100644 --- a/codex-rs/tui/src/app/tests/safety_buffering.rs +++ b/codex-rs/tui/src/app/tests/safety_buffering.rs @@ -747,10 +747,10 @@ goals = true let expected_source_tokens = if committed_steer.is_some() { 150 } else { 50 }; assert_eq!(source_goal.objective, RETRY_GOAL); assert_eq!(source_goal.tokens_used, expected_source_tokens); - assert_eq!(source_goal.time_used_seconds, 12); + assert!(source_goal.time_used_seconds >= 12); assert_eq!(retry_goal.objective, RETRY_GOAL); assert!(retry_goal.tokens_used >= expected_source_tokens); - assert!(retry_goal.time_used_seconds >= 12); + assert!(retry_goal.time_used_seconds >= source_goal.time_used_seconds); let request_bodies = server .requests() diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index a1a8a8cc3f..69358321f1 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -198,6 +198,7 @@ impl ChatWidget { | ServerNotification::AccountRateLimitsUpdated(_) | ServerNotification::ThreadStarted(_) | ServerNotification::ThreadStatusChanged(_) + | ServerNotification::ThreadReverted(_) | ServerNotification::ThreadArchived(_) | ServerNotification::ThreadDeleted(_) | ServerNotification::ThreadUnarchived(_)